Flush telemetry in a detached subprocess (#275)

Author: carole-lavillonniere

cmd/flush_telemetry.go

@@ -0,0 +1,43 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/localstack/lstk/internal/env"
+	"github.com/localstack/lstk/internal/log"
+	"github.com/localstack/lstk/internal/telemetry"
+	"github.com/localstack/lstk/internal/tracing"
+)
+
+// runFlushTelemetry handles the flusher subprocess. It bypasses the normal
+// Execute() boot path — no logger, keyring, telemetry client, or cobra tree —
+// so it writes nothing to disk and cannot spawn another flusher.
+func runFlushTelemetry(ctx context.Context, args []string) error {
+	cfg := env.Init()
+	if cfg.TracesEnabled {
+		shutdown := tracing.Init(ctx, log.Nop())
+		defer func() {
+			// Fresh context: ctx may be cancelled and Shutdown would skip the flush.
+			shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			_ = shutdown(shutCtx)
+		}()
+	}
+
+	endpoint := ""
+	for i := 0; i < len(args); i++ {
+		if args[i] == "--endpoint" && i+1 < len(args) {
+			endpoint = args[i+1]
+			i++
+		}
+	}
+	if endpoint == "" {
+		return fmt.Errorf("missing --endpoint")
+	}
+
+	ctx = tracing.ContextWithRemoteParent(ctx, os.Getenv)
+	return telemetry.RunFlush(ctx, endpoint, os.Stdin)
+}

cmd/root.go

@@ -93,6 +93,10 @@ func NewRootCmd(cfg *env.Env, tel *telemetry.Client, logger log.Logger) *cobra.C
 }
 
 func Execute(ctx context.Context) error {
+	if len(os.Args) > 1 && os.Args[1] == telemetry.FlushCommandName {
+		return runFlushTelemetry(ctx, os.Args[2:])
+	}
+
 	cfg := env.Init()
 
 	logger, cleanup, err := newLogger()

go.mod

@@ -26,7 +26,9 @@ require (
 	go.opentelemetry.io/otel v1.44.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0
 	go.opentelemetry.io/otel/sdk v1.44.0
+	go.opentelemetry.io/otel/trace v1.44.0
 	go.uber.org/mock v0.6.0
+	golang.org/x/sys v0.45.0
 	golang.org/x/term v0.43.0
 	gopkg.in/ini.v1 v1.67.2
 	gotest.tools/v3 v3.5.2
@@ -82,11 +84,9 @@ require (
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect
 	go.opentelemetry.io/otel/metric v1.44.0 // indirect
-	go.opentelemetry.io/otel/trace v1.44.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.10.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/net v0.55.0 // indirect
-	golang.org/x/sys v0.45.0 // indirect
 	golang.org/x/text v0.37.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect

internal/container/telemetry_test.go

@@ -33,7 +33,7 @@ func newCapturingTelClient(t *testing.T) (*telemetry.Client, <-chan map[string]a
 		w.WriteHeader(http.StatusOK)
 	}))
 	t.Cleanup(srv.Close)
-	return telemetry.New(srv.URL, false), ch
+	return telemetry.NewWithInProcessFlush(srv.URL), ch
 }
 
 func TestStop_EmitsLifecycleStopEvent(t *testing.T) {

internal/telemetry/client.go

@@ -1,37 +1,30 @@
 package telemetry
 
 import (
-	"bytes"
 	"context"
-	"encoding/json"
-	"fmt"
-	"net/http"
 	"os"
 	"runtime"
 	"sync"
 	"time"
 
 	"github.com/google/uuid"
-	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
-
-	"github.com/localstack/lstk/internal/version"
 )
 
-func userAgent() string {
-	return fmt.Sprintf("localstack lstk/%s (%s; %s)", version.Version(), runtime.GOOS, runtime.GOARCH)
-}
+// pendingCap bounds in-memory events; on overflow the oldest is dropped.
+const pendingCap = 64
 
 type Client struct {
 	enabled   bool
 	sessionID string
 	machineID string
 	authToken string
 
-	httpClient *http.Client
-	endpoint   string
+	endpoint string
+	flushFn  func(ctx context.Context, endpoint string, events []eventBody)
 
-	events        chan eventBody
-	done          chan struct{}
+	mu            sync.Mutex
+	pending       []eventBody
+	traceCtx      context.Context // last Emit ctx; carries the command span for trace propagation
 	closeOnce     sync.Once
 	machineIDOnce sync.Once
 }
@@ -46,37 +39,23 @@ func New(endpoint string, disabled bool) *Client {
 	if disabled {
 		return &Client{enabled: false}
 	}
-	c := &Client{
+	return &Client{
 		enabled:   true,
 		sessionID: uuid.NewString(),
-		// http.Client has no default timeout (zero means none). Without one, a
-		// slow or unreachable endpoint would block the worker goroutine.
-		httpClient: &http.Client{
-			Timeout: 3 * time.Second,
-			Transport: otelhttp.NewTransport(
-				http.DefaultTransport,
-				otelhttp.WithSpanNameFormatter(func(_ string, r *http.Request) string {
-					return "telemetry " + r.Method + " " + r.URL.Path
-				}),
-			),
-		},
-		endpoint: endpoint,
-		events:   make(chan eventBody, 64),
-		done:     make(chan struct{}),
+		endpoint:  endpoint,
+		flushFn:   spawnDetachedFlusher,
+		pending:   make([]eventBody, 0, pendingCap),
 	}
-	go c.worker()
-	return c
 }
 
 type requestBody struct {
 	Events []eventBody `json:"events"`
 }
 
 type eventBody struct {
-	ctx      context.Context // not serialized; carries context to the worker
-	Name     string          `json:"name"`
-	Metadata eventMetadata   `json:"metadata"`
-	Payload  any             `json:"payload"`
+	Name     string        `json:"name"`
+	Metadata eventMetadata `json:"metadata"`
+	Payload  any           `json:"payload"`
 }
 
 type eventMetadata struct {
@@ -101,56 +80,41 @@ func (c *Client) Emit(ctx context.Context, name string, payload map[string]any)
 	}
 
 	body := eventBody{
-		ctx:  context.WithoutCancel(ctx),
 		Name: name,
 		Metadata: eventMetadata{
 			ClientTime: time.Now().UTC().Format("2006-01-02 15:04:05.000000"),
 			SessionID:  c.sessionID,
 		},
 		Payload: enriched,
 	}
-	select {
-	case c.events <- body:
-	default:
-	}
-}
-
-func (c *Client) worker() {
-	defer close(c.done)
-	for body := range c.events {
-		c.send(body)
-	}
-}
 
-func (c *Client) send(body eventBody) {
-	data, err := json.Marshal(requestBody{Events: []eventBody{body}})
-	if err != nil {
-		return
-	}
-
-	req, err := http.NewRequestWithContext(body.ctx, http.MethodPost, c.endpoint, bytes.NewReader(data))
-	if err != nil {
-		return
-	}
-	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("User-Agent", userAgent())
-
-	resp, err := c.httpClient.Do(req)
-	if err != nil {
-		return
+	c.mu.Lock()
+	if len(c.pending) >= pendingCap {
+		c.pending = c.pending[1:]
 	}
-	_ = resp.Body.Close()
+	c.pending = append(c.pending, body)
+	c.traceCtx = context.WithoutCancel(ctx)
+	c.mu.Unlock()
 }
 
-// Close stops accepting new events, drains the event buffer, and blocks until
-// all pending HTTP requests have completed. Call it before process exit to
-// avoid dropping telemetry events.
+// Close hands pending events to a detached subprocess and returns immediately,
+// so analytics endpoint latency never delays command exit.
 func (c *Client) Close() {
 	if !c.enabled {
 		return
 	}
 	c.closeOnce.Do(func() {
-		close(c.events)
-		<-c.done
+		c.mu.Lock()
+		pending := c.pending
+		traceCtx := c.traceCtx
+		c.pending = nil
+		c.mu.Unlock()
+		if len(pending) == 0 {
+			return
+		}
+		if traceCtx == nil {
+			traceCtx = context.Background()
+		}
+		c.flushFn(traceCtx, c.endpoint, pending)
 	})
 }