[POA-2544] Updates to the initial telemetry workflow (#65)

With this PR, we have made some changes to the initial telemetry flows. 
1. We have removed the statsLogDelay config. We now no longer send an
initial telemetry after 60 seconds of startup.
2. We now report successful telemetry whenever we see a successful
capture of a request and a response. This compensates for the long wait
time of 5 minutes, i.e., the default telemetry interval.

Implementation inspired from
[here](https://postmanlabs.atlassian.net/wiki/spaces/PO/pages/5811765834/Plan+Warning+about+not+seeing+HTTP+traffic+needs+to+be+adjusted?focusedCommentId=5830148712).

Author: shreys7

apidump/apidump.go

@@ -17,7 +17,6 @@ import (
 	kgxapi "github.com/akitasoftware/akita-libs/api_schema"
 	"github.com/akitasoftware/akita-libs/buffer_pool"
 	"github.com/akitasoftware/akita-libs/tags"
-	"github.com/akitasoftware/go-utils/math"
 	"github.com/akitasoftware/go-utils/optionals"
 	"github.com/pkg/errors"
 	"github.com/postmanlabs/postman-insights-agent/apispec"
@@ -116,9 +115,6 @@ type Args struct {
 	// How often to rotate learn sessions; set to zero to disable rotation.
 	LearnSessionLifetime time.Duration
 
-	// Print packet capture statistics after N seconds.
-	StatsLogDelay int
-
 	// Periodically report telemetry every N seconds thereafter
 	TelemetryInterval int
 
@@ -153,15 +149,20 @@ type apidump struct {
 	backendSvcName string
 	learnClient    rest.LearnClient
 
-	startTime   time.Time
-	dumpSummary *Summary
+	startTime        time.Time
+	dumpSummary      *Summary
+	successTelemetry *trace.SuccessTelemetry
 }
 
 // Start a new apidump session based on the given arguments.
 func newSession(args *Args) *apidump {
 	a := &apidump{
 		Args:      args,
 		startTime: time.Now(),
+		successTelemetry: &trace.SuccessTelemetry{
+			Channel: make(chan struct{}),
+			Once:    sync.Once{},
+		},
 	}
 	return a
 }
@@ -200,7 +201,7 @@ func (a *apidump) LookupService() error {
 	return nil
 }
 
-// Send the initial mesage to the backend indicating successful start
+// Send the initial message to the backend indicating successful start
 func (a *apidump) SendInitialTelemetry() {
 	// Do not send packet capture telemetry for local captures.
 	if !a.TargetIsRemote() {
@@ -209,11 +210,11 @@ func (a *apidump) SendInitialTelemetry() {
 
 	// XXX(cns):  The observed duration serves as a key for upserting packet
 	//    telemetry, so it needs to be the same here as in the packet
-	//    telemetry sent sixty seconds after startup.
+	//    telemetry sent 5 minutes after startup.
 	req := kgxapi.PostInitialClientTelemetryRequest{
 		ClientID:                  a.ClientID,
 		ObservedStartingAt:        a.startTime,
-		ObservedDurationInSeconds: a.StatsLogDelay,
+		ObservedDurationInSeconds: a.TelemetryInterval,
 		SendsWitnessPayloads:      a.ReproMode,
 		CLIVersion:                version.ReleaseVersion().String(),
 		CLITargetArch:             architecture.GetCanonicalArch(),
@@ -234,7 +235,7 @@ func (a *apidump) SendInitialTelemetry() {
 // Send a message to the backend indicating failure to start and a cause
 func (a *apidump) SendErrorTelemetry(errorType api_schema.ApidumpErrorType, err error) {
 	req := &kgxapi.PostClientPacketCaptureStatsRequest{
-		ObservedDurationInSeconds: a.StatsLogDelay,
+		ObservedDurationInSeconds: a.TelemetryInterval,
 		ApidumpError:              errorType,
 		ApidumpErrorText:          err.Error(),
 	}
@@ -414,31 +415,16 @@ func (a *apidump) RotateLearnSession(done <-chan struct{}, collectors []trace.Le
 
 // Goroutine to send telemetry, stop when "done" is closed.
 //
-// Prints a summary after a short delay.  This ensures that statistics will
-// appear in customer logs close to when the process is started.
-// Omits if args.StatsLogDelay is <= 0.
-//
 // Sends telemetry to the server on a regular basis.
 // Omits if args.TelemetryInterval is <= 0
 func (a *apidump) TelemetryWorker(done <-chan struct{}) {
-	if a.StatsLogDelay <= 0 && a.TelemetryInterval <= 0 {
+	if a.TelemetryInterval <= 0 {
 		return
 	}
 
 	a.SendInitialTelemetry()
 
-	if a.StatsLogDelay > 0 {
-		// Wait while capturing statistics.
-		time.Sleep(time.Duration(a.StatsLogDelay) * time.Second)
-
-		// Print telemetry data.
-		printer.Stderr.Infof("Printing packet capture statistics after %d seconds of capture.\n", a.StatsLogDelay)
-		a.dumpSummary.PrintPacketCounts()
-		a.dumpSummary.PrintWarnings()
-
-		a.SendPacketTelemetry(a.StatsLogDelay)
-	}
-
+	subsequentTelemetrySent := false
 	if a.TelemetryInterval > 0 {
 		ticker := time.NewTicker(time.Duration(a.TelemetryInterval) * time.Second)
 
@@ -449,6 +435,11 @@ func (a *apidump) TelemetryWorker(done <-chan struct{}) {
 			case now := <-ticker.C:
 				duration := int(now.Sub(a.startTime) / time.Second)
 				a.SendPacketTelemetry(duration)
+				subsequentTelemetrySent = true
+			case <-a.successTelemetry.Channel:
+				if !subsequentTelemetrySent {
+					a.SendPacketTelemetry(a.TelemetryInterval)
+				}
 			}
 		}
 	}
@@ -641,15 +632,8 @@ func (a *apidump) Run() error {
 	// when the main collection process does.
 	if a.TargetIsRemote() {
 		{
-			// Record the first resource usage data slightly before the
-			// stats log delay to ensure we include usage data in the first
-			// telemetry upload.
-			var delay time.Duration
-			if 0 < a.StatsLogDelay {
-				delay = time.Duration(math.Max(a.StatsLogDelay-5, 1)) * time.Second
-			}
-
-			go usage.Poll(stop, delay, time.Duration(a.ProcFSPollingInterval)*time.Second)
+			// Record the first usage immediately (sending delay = 0) since we want to include it in the success telemetry
+			go usage.Poll(stop, 0, time.Duration(a.ProcFSPollingInterval)*time.Second)
 		}
 
 		go a.TelemetryWorker(stop)
@@ -724,8 +708,9 @@ func (a *apidump) Run() error {
 			// trace is empty or not.)  In the future we could add columns for both
 			// pre- and post-filtering.
 			collector = &trace.PacketCountCollector{
-				PacketCounts: summary,
-				Collector:    collector,
+				PacketCounts:     summary,
+				Collector:        collector,
+				SuccessTelemetry: a.successTelemetry,
 			}
 
 			// Subsampling.

apispec/defaults.go

@@ -27,9 +27,6 @@ const (
 	// How many requests to capture per minute.
 	DefaultRateLimit = 1000.0
 
-	// How long to wait after starting up before printing packet-capture statistics.
-	DefaultStatsLogDelay_seconds = 60
-
 	// How often to upload client telemetry.
 	DefaultTelemetryInterval_seconds = 5 * 60 // 5 minutes
 

cmd/internal/apidump/apidump.go

@@ -32,7 +32,6 @@ var (
 	execCommandUserFlag     string
 	pluginsFlag             []string
 	traceRotateFlag         string
-	statsLogDelay           int
 	telemetryInterval       int
 	procFSPollingInterval   int
 	collectTCPAndTLSReports bool
@@ -212,7 +211,6 @@ func apidumpRunInternal(cmd *cobra.Command, _ []string) error {
 		ExecCommandUser:         execCommandUserFlag,
 		Plugins:                 plugins,
 		LearnSessionLifetime:    traceRotateInterval,
-		StatsLogDelay:           statsLogDelay,
 		TelemetryInterval:       telemetryInterval,
 		ProcFSPollingInterval:   procFSPollingInterval,
 		CollectTCPAndTLSReports: collectTCPAndTLSReports,
@@ -309,13 +307,6 @@ func init() {
 	)
 	Cmd.Flags().MarkHidden("trace-rotate")
 
-	Cmd.Flags().IntVar(
-		&statsLogDelay,
-		"stats-log-delay",
-		apispec.DefaultStatsLogDelay_seconds,
-		"Print packet capture statistics after N seconds.",
-	)
-
 	Cmd.Flags().IntVar(
 		&telemetryInterval,
 		"telemetry-interval",

trace/collector.go

@@ -4,6 +4,7 @@ import (
 	"math"
 	"sort"
 	"strconv"
+	"sync"
 
 	"github.com/OneOfOne/xxhash"
 	"github.com/akitasoftware/akita-libs/akid"
@@ -14,6 +15,11 @@ import (
 	"github.com/spf13/viper"
 )
 
+type SuccessTelemetry struct {
+	Channel chan struct{}
+	Once    sync.Once
+}
+
 type Collector interface {
 	// Hands new data from network to the collector. The implementation may choose
 	// to process them asynchronously (e.g. to wait for the response to a
@@ -98,8 +104,9 @@ func (sc *UserTrafficCollector) Close() error {
 
 // This is a shim to add packet counts based on payload type.
 type PacketCountCollector struct {
-	PacketCounts PacketCountConsumer
-	Collector    Collector
+	PacketCounts     PacketCountConsumer
+	Collector        Collector
+	SuccessTelemetry *SuccessTelemetry
 }
 
 // Don't record self-generated traffic in the breakdown by hostname,
@@ -192,9 +199,17 @@ func (pc *PacketCountCollector) Process(t akinet.ParsedNetworkTraffic) error {
 			Unparsed:  1,
 		})
 	}
+	if pc.PacketCounts.HasRequestAndResponse() {
+		pc.SendSuccessTelemetry()
+	}
 	return pc.Collector.Process(t)
 }
 
+func (pc *PacketCountCollector) SendSuccessTelemetry() {
+	pc.SuccessTelemetry.Once.Do(func() {
+		pc.SuccessTelemetry.Channel <- struct{}{}
+	})
+}
 func (pc *PacketCountCollector) Close() error {
 	return pc.Collector.Close()
 }

trace/stats.go

@@ -15,6 +15,9 @@ import (
 type PacketCountConsumer interface {
 	// Add an additional measurement to the current count
 	Update(delta PacketCounts)
+
+	// Returns true if we have successfully received at least one request and a response
+	HasRequestAndResponse() bool
 }
 
 // Discard the count
@@ -127,6 +130,12 @@ func (s *PacketCounter) Update(c PacketCounts) {
 	s.total.Add(c)
 }
 
+func (s *PacketCounter) HasRequestAndResponse() bool {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+	return s.total.HTTPRequests > 0 && s.total.HTTPResponses > 0
+}
+
 func (s *PacketCounter) Total() PacketCounts {
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()