milo: fix step duration

client/logdog/annotee/processor.go

 // Copyright 2015 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 annotee
 
 import (
         "bufio"
         "bytes"
         "fmt"
@@ skipping 85 matching lines @@
         Execution *annotation.Execution
 
         // MetadataUpdateInterval is the amount of time to wait after stream metadata
         // updates to push the updated metadata protobuf to the metadata stream.
         //
         //      - If this is < 0, metadata will only be pushed at the beginning and end of
         //        a step.
         //      - If this equals 0, metadata will be pushed every time it's updated.
         //      - If this is 0, DefaultMetadataUpdateInterval will be used.
         MetadataUpdateInterval time.Duration
+
+        // Offline specifies whether parsing happens not at the same time as
+        // emitting. If true and CURRENT_TIMESTAMP annotations are not provided
+        // then step start/end times are left empty.
+        Offline bool
 }
 
 // Processor consumes data from a list of Stream entries and interacts with the
 // supplied Client instance.
 //
 // A Processor must be instantiated with New.
 type Processor struct {
         ctx context.Context
         o   *Options
 
         astate       *annotation.State
         stepHandlers map[string]*stepHandler
 }
 
 // New instantiates a new Processor.
 func New(c context.Context, o Options) *Processor {
         p := Processor{
                 ctx: c,
                 o:   &o,
 
                 stepHandlers: make(map[string]*stepHandler),
         }
         p.astate = &annotation.State{
                 LogNameBase: o.Base,
                 Callbacks:   &annotationCallbacks{&p},
                 Execution:   o.Execution,
                 Clock:       clock.Get(c),
+                Offline:     o.Offline,
         }
         return &p
 }
 
 // RunStreams executes the Processor, consuming data from its configured streams
 // and forwarding it to LogDog. Run will block until all streams have
 // terminated.
 //
 // If a stream terminates with an error, or if there is an error processing the
 // stream data, Run will return an error. If multiple Streams fail with errors,
@@ skipping 304 matching lines @@
                         sendLatest()
                 }
         }
 }
 
 func (h *stepHandler) finish() {
         if h.closed {
                 return
         }
 
-        // Close the handler. This may send one last annotation to summarize the
+        // This may send one last annotation to summarize the
         // state if closing changed it.
-        if h.step.Close() {
+        if h.step.Close(nil) {
                 // Manually mark it updated, since Close callbacks will have unregistered
                 // us from the standard Updated() reporting loop.
                 h.updated()
         }
 
         // Close and reap our meter goroutine.
         close(h.annotationC)
         <-h.annotationFinishedC
 
         // Close all streams associated with this handler.
@@ skipping 149 matching lines @@
                 return ""
         }
         return strings.TrimSpace(line[3 : len(line)-3])
 }
 
 func buildAnnotation(name string, params ...string) string {
         v := make([]string, 1, 1+len(params))
         v[0] = name
         return "@@@" + strings.Join(append(v, params...), "@") + "@@@"
 }