LogDog: BigTable batching schema.

server/cmd/logdog_collector/main.go

 // Copyright 2016 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package main
 
 import (
         "fmt"
         "time"
 
@@ skipping 69 matching lines @@
         if !exists {
                 log.Fields{
                         "subscription": sub,
                 }.Errorf(c, "Subscription does not exist.")
                 return errInvalidConfig
         }
         log.Fields{
                 "subscription": sub,
         }.Infof(c, "Successfully validated Pub/Sub subscription.")
 
-        // Initialize our Storage.
-        s, err := a.IntermediateStorage(c)
+        st, err := a.IntermediateStorage(c)
         if err != nil {
-                log.WithError(err).Errorf(c, "Failed to get storage instance.")
                 return err
         }
-        defer s.Close()
+        defer st.Close()
 
         // Application shutdown will now operate by cancelling the Collector's
         // shutdown Context.
         shutdownCtx, shutdownFunc := context.WithCancel(c)
         a.SetShutdownFunc(shutdownFunc)
 
         // Start an ACK buffer so that we can batch ACKs. Note that we do NOT use the
         // shutdown context here, as we want clean shutdowns to continue to ack any
         // buffered messages.
         ab := ackbuffer.New(c, ackbuffer.Config{
                 Ack: ackbuffer.NewACK(ps, sub, 0),
         })
         defer ab.CloseAndFlush()
 
         // Initialize our Collector service object using a caching Coordinator
         // interface.
         coord := coordinator.NewCoordinator(a.Coordinator())
         coord = coordinator.NewCache(coord, int(ccfg.StateCacheSize), ccfg.StateCacheExpiration.Duration())
 
         coll := collector.Collector{
-                Coordinator:        coord,
-                Storage:            s,
-                MaxParallelBundles: int(ccfg.Workers),
-                MaxIngestWorkers:   int(ccfg.Workers),
+                Coordinator:       coord,
+                Storage:           st,
+                MaxMessageWorkers: int(ccfg.MaxMessageWorkers),
         }
         defer coll.Close()
 
         // Execute our main Subscriber loop. It will run until the supplied Context
         // is cancelled.
         clk := clock.Get(c)
         engine := subscriber.Subscriber{
-                S: subscriber.NewSource(ps, sub, 0),
-                A: ab,
-
-                PullWorkers:    int(ccfg.TransportWorkers),
-                HandlerWorkers: int(ccfg.Workers),
+                S:       subscriber.NewSource(ps, sub),
+                A:       ab,
+                Workers: int(ccfg.MaxConcurrentMessages),
         }
         engine.Run(shutdownCtx, func(msg *pubsub.Message) bool {
                 c := log.SetField(c, "messageID", msg.ID)
                 log.Fields{
                         "ackID": msg.AckID,
                         "size":  len(msg.Data),
                 }.Infof(c, "Received Pub/Sub Message.")
 
                 startTime := clk.Now()
                 err := coll.Process(c, msg.Data)
@@ skipping 29 matching lines @@
 
         log.Debugf(c, "Collector finished.")
         return nil
 }
 
 // Entry point.
 func main() {
         a := application{}
         a.Run(context.Background(), a.runCollector)
 }