feat: add request ID to Tail/Subscribe calls (#1388)

* feat: add request ID to Tail/Subscribe calls

* chore: use canonical header names

Author: lionello

src/pkg/auth/interceptor.go

@@ -7,7 +7,7 @@ import (
 	"github.com/bufbuild/connect-go"
 )
 
-const XDefangOrgID = "x-defang-orgid"
+const XDefangOrgID = "X-Defang-Orgid"
 
 type authInterceptor struct {
 	authorization string
@@ -20,8 +20,8 @@ func NewAuthInterceptor(token, orgID string) connect.Interceptor {
 
 func (a *authInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
 	return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
-		req.Header().Set("authorization", a.authorization)
-		req.Header().Set("content-type", "application/grpc") // same as the gRPC client
+		req.Header().Set("Authorization", a.authorization)
+		req.Header().Set("Content-Type", "application/grpc") // same as the gRPC client
 		req.Header().Set(XDefangOrgID, a.orgID)
 		return next(ctx, req)
 	}
@@ -30,8 +30,8 @@ func (a *authInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
 func (a *authInterceptor) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
 	return func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn {
 		conn := next(ctx, spec)
-		conn.RequestHeader().Set("authorization", a.authorization)
-		conn.RequestHeader().Set("content-type", "application/grpc") // same as the gRPC client
+		conn.RequestHeader().Set("Authorization", a.authorization)
+		conn.RequestHeader().Set("Content-Type", "application/grpc") // same as the gRPC client
 		conn.RequestHeader().Set(XDefangOrgID, a.orgID)
 		return conn
 	}

src/pkg/cli/client/grpc_logger.go

@@ -40,8 +40,20 @@ func (g grpcLogger) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
 	}
 }
 
-func (grpcLogger) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
-	return next
+func (g grpcLogger) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
+	return func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn {
+		conn := next(ctx, spec)
+
+		// Add a request ID to the context
+		requestId := pkg.RandomID()
+		conn.RequestHeader().Add("X-Request-Id", requestId)
+
+		// Get the request type name
+		reqType := spec.Procedure
+
+		term.Debug(g.prefix, requestId, reqType, "streaming connection established")
+		return conn
+	}
 }
 
 func (grpcLogger) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {