Allow streaming to client

mxpv · mxpv · commit 17b6e1ef8509 · 2025-04-23T18:03:29.000-07:00
Signed-off-by: Maksym Pavlenko &lt;pavlenko.maksym@gmail.com&gt;
diff --git a/api/next.pb.txt b/api/next.pb.txt
@@ -7779,15 +7779,16 @@ file {
       name: "http_debug"
       number: 5
       label: LABEL_OPTIONAL
-      type: TYPE_BOOL
+      type: TYPE_ENUM
+      type_name: ".containerd.types.transfer.HTTPDebug"
       json_name: "httpDebug"
     }
     field {
-      name: "http_trace"
+      name: "logs_stream"
       number: 6
       label: LABEL_OPTIONAL
-      type: TYPE_BOOL
-      json_name: "httpTrace"
+      type: TYPE_STRING
+      json_name: "logsStream"
     }
     nested_type {
       name: "HeadersEntry"
@@ -7867,6 +7868,25 @@ file {
       json_name: "expireAt"
     }
   }
+  enum_type {
+    name: "HTTPDebug"
+    value {
+      name: "DISABLED"
+      number: 0
+    }
+    value {
+      name: "DEBUG"
+      number: 1
+    }
+    value {
+      name: "TRACE"
+      number: 2
+    }
+    value {
+      name: "BOTH"
+      number: 3
+    }
+  }
   enum_type {
     name: "AuthType"
     value {
diff --git a/api/types/transfer/registry.pb.go b/api/types/transfer/registry.pb.go
diff --git a/api/types/transfer/registry.proto b/api/types/transfer/registry.proto
@@ -27,6 +27,16 @@ message OCIRegistry {
 	RegistryResolver resolver = 2;
 }
 
+enum HTTPDebug {
+	DISABLED = 0;
+	// Enable HTTP debugging
+	DEBUG = 1;
+	// Enable HTTP requests tracing
+	TRACE = 2;
+	// Enable both HTTP debugging and requests tracing
+	BOTH = 3;
+}
+
 message RegistryResolver {
 	// auth_stream is used to refer to a stream which auth callbacks may be
 	// made on.
@@ -41,11 +51,12 @@ message RegistryResolver {
 	// Force skip verify
 	// CA callback? Client TLS callback?
 
-	// Enable HTTP debugging
-	bool http_debug = 5;
+	// Whether to debug/trace HTTP requests to OCI registry.
+	HTTPDebug http_debug = 5;
 
-	// Enable HTTP requests tracing
-	bool http_trace = 6;
+	// Stream ID to use for HTTP logs (when logs are streamed to client).
+	// When empty, logs are written to containerd logs.
+	string logs_stream = 6;
 }
 
 // AuthRequest is sent as a callback on a stream
diff --git a/cmd/ctr/commands/images/pull.go b/cmd/ctr/commands/images/pull.go
@@ -159,11 +159,12 @@ command. As part of this process, we do the following:
 			if cliContext.Bool("plain-http") {
 				opts = append(opts, registry.WithDefaultScheme("http"))
 			}
+			logStream := log.G(ctx).Writer()
 			if cliContext.Bool("http-dump") {
-				opts = append(opts, registry.WithHTTPDebug())
+				opts = append(opts, registry.WithHTTPDebug(), registry.WithClientStream(logStream))
 			}
 			if cliContext.Bool("http-trace") {
-				opts = append(opts, registry.WithHTTPTrace())
+				opts = append(opts, registry.WithHTTPTrace(), registry.WithClientStream(logStream))
 			}
 			reg, err := registry.NewOCIRegistry(ctx, ref, opts...)
 			if err != nil {
diff --git a/cmd/ctr/commands/resolver.go b/cmd/ctr/commands/resolver.go
@@ -101,7 +101,7 @@ func GetResolver(ctx context.Context, cliContext *cli.Context) (remotes.Resolver
 
 	if cliContext.Bool("http-dump") {
 		hostOptions.UpdateClient = func(client *http.Client) error {
-			httpdbg.DumpRequests(ctx, client)
+			httpdbg.DumpRequests(ctx, client, nil)
 			return nil
 		}
 	}
diff --git a/core/transfer/registry/registry.go b/core/transfer/registry/registry.go
@@ -51,6 +51,7 @@ type registryOpts struct {
 	defaultScheme string
 	httpDebug     bool
 	httpTrace     bool
+	localStream   io.WriteCloser
 }
 
 // Opt sets registry-related configurations.
@@ -104,6 +105,15 @@ func WithHTTPTrace() Opt {
 	}
 }
 
+// WithClientStream tells the registry to stream HTTP debug data back to the client.
+// Applicable only when HTTP debug or tracing enabled.
+func WithClientStream(writer io.WriteCloser) Opt {
+	return func(o *registryOpts) error {
+		o.localStream = writer
+		return nil
+	}
+}
+
 // NewOCIRegistry initializes with hosts, authorizer callback, and headers
 func NewOCIRegistry(ctx context.Context, ref string, opts ...Opt) (*OCIRegistry, error) {
 	var ropts registryOpts
@@ -134,7 +144,7 @@ func NewOCIRegistry(ctx context.Context, ref string, opts ...Opt) (*OCIRegistry,
 
 	hostOptions.UpdateClient = func(client *http.Client) error {
 		if ropts.httpDebug {
-			httpdbg.DumpRequests(ctx, client)
+			httpdbg.DumpRequests(ctx, client, ropts.localStream)
 		}
 		if ropts.httpTrace {
 			httpdbg.DumpTraces(ctx, client)
@@ -156,6 +166,7 @@ func NewOCIRegistry(ctx context.Context, ref string, opts ...Opt) (*OCIRegistry,
 		defaultScheme: ropts.defaultScheme,
 		httpDebug:     ropts.httpDebug,
 		httpTrace:     ropts.httpTrace,
+		localStream:   ropts.localStream,
 	}, nil
 }
 
@@ -184,8 +195,9 @@ type OCIRegistry struct {
 
 	defaultScheme string
 
-	httpDebug bool
-	httpTrace bool
+	httpDebug   bool
+	httpTrace   bool
+	localStream io.WriteCloser
 
 	// This could be an interface which returns resolver?
 	// Resolver could also be a plug-able interface, to call out to a program to fetch?
@@ -285,10 +297,43 @@ func (r *OCIRegistry) MarshalAny(ctx context.Context, sm streaming.StreamCreator
 		}()
 		res.AuthStream = sid
 	}
+
+	if r.httpDebug || r.httpTrace {
+		switch {
+		case r.httpDebug && r.httpTrace:
+			res.HttpDebug = transfertypes.HTTPDebug_BOTH
+		case r.httpDebug:
+			res.HttpDebug = transfertypes.HTTPDebug_DEBUG
+		case r.httpTrace:
+			res.HttpDebug = transfertypes.HTTPDebug_TRACE
+		default:
+			res.HttpDebug = transfertypes.HTTPDebug_DISABLED
+		}
+
+		if r.localStream != nil {
+			res.LogsStream = tstreaming.GenerateID("http-debug-logs")
+
+			stream, err := sm.Create(ctx, res.LogsStream)
+			if err != nil {
+				return nil, fmt.Errorf("failed to create stream for HTTP debug logs: %w", err)
+			}
+
+			go func() {
+				// Start pumping logs to the client
+				_, err := io.Copy(r.localStream, tstreaming.ReceiveStream(ctx, stream))
+				if err != nil && !errors.Is(err, io.EOF) {
+					log.G(ctx).WithError(err).Error("failed to copy HTTP debug logs stream")
+				}
+
+				if err := r.localStream.Close(); err != nil {
+					log.G(ctx).WithError(err).Error("failed to close HTTP debug logs local stream")
+				}
+			}()
+		}
+	}
+
 	res.HostDir = r.hostDir
 	res.DefaultScheme = r.defaultScheme
-	res.HttpDebug = r.httpDebug
-	res.HttpTrace = r.httpTrace
 	s := &transfertypes.OCIRegistry{
 		Reference: r.reference,
 		Resolver:  res,
@@ -334,18 +379,41 @@ func (r *OCIRegistry) UnmarshalAny(ctx context.Context, sm streaming.StreamGette
 			r.headers.Add(k, v)
 		}
 
-		r.httpDebug = s.Resolver.GetHttpDebug()
-		r.httpTrace = s.Resolver.GetHttpTrace()
-	}
+		if s.Resolver.HttpDebug != transfertypes.HTTPDebug_DISABLED {
+			var writer io.WriteCloser
 
-	hostOptions.UpdateClient = func(client *http.Client) error {
-		if r.httpDebug {
-			httpdbg.DumpRequests(ctx, client)
-		}
-		if r.httpTrace {
-			httpdbg.DumpTraces(ctx, client)
+			// Stream to local client.
+			if sid := s.Resolver.LogsStream; sid != "" {
+				stream, err := sm.Get(ctx, sid)
+				if err != nil {
+					return fmt.Errorf("failed to get stream for HTTP debug logs: %w", err)
+				}
+
+				writer = tstreaming.WriteByteStream(ctx, stream)
+			} else {
+				writer = log.G(ctx).Writer()
+			}
+
+			go func() {
+				<-ctx.Done()
+				if err := writer.Close(); err != nil {
+					log.G(ctx).Errorf("failed to close HTTP debug logs stream: %v", err)
+				}
+			}()
+
+			hostOptions.UpdateClient = func(client *http.Client) error {
+				switch s.Resolver.HttpDebug {
+				case transfertypes.HTTPDebug_DEBUG:
+					httpdbg.DumpRequests(ctx, client, writer)
+				case transfertypes.HTTPDebug_TRACE:
+					httpdbg.DumpTraces(ctx, client)
+				case transfertypes.HTTPDebug_BOTH:
+					httpdbg.DumpRequests(ctx, client, writer)
+					httpdbg.DumpTraces(ctx, client)
+				}
+				return nil
+			}
 		}
-		return nil
 	}
 
 	r.reference = s.Reference
diff --git a/pkg/httpdbg/debug.go b/pkg/httpdbg/debug.go
@@ -62,10 +62,14 @@ func (t debugTransport) RoundTrip(req *http.Request) (*http.Response, error) {
 }
 
 // DumpRequests wraps the underlying http.Client transport to logs all requests/responses to the log.
-func DumpRequests(ctx context.Context, client *http.Client) {
+func DumpRequests(ctx context.Context, client *http.Client, writer io.Writer) {
+	if writer == nil {
+		writer = log.G(ctx).Writer()
+	}
+
 	client.Transport = debugTransport{
 		transport: client.Transport,
-		writer:    log.G(ctx).Writer(),
+		writer:    writer,
 	}
 }
 
diff --git a/vendor/github.com/containerd/containerd/api/types/transfer/registry.pb.go b/vendor/github.com/containerd/containerd/api/types/transfer/registry.pb.go
diff --git a/vendor/github.com/containerd/containerd/api/types/transfer/registry.proto b/vendor/github.com/containerd/containerd/api/types/transfer/registry.proto

Original file line number	Diff line number	Diff line change
`@@ -159,11 +159,12 @@ command. As part of this process, we do the following:`
`159`	`159`	`if cliContext.Bool("plain-http") {`
`160`	`160`	`opts = append(opts, registry.WithDefaultScheme("http"))`
`161`	`161`	`}`
	`162`	`+ logStream := log.G(ctx).Writer()`
`162`	`163`	`if cliContext.Bool("http-dump") {`
`163`		`- opts = append(opts, registry.WithHTTPDebug())`
	`164`	`+ opts = append(opts, registry.WithHTTPDebug(), registry.WithClientStream(logStream))`
`164`	`165`	`}`
`165`	`166`	`if cliContext.Bool("http-trace") {`
`166`		`- opts = append(opts, registry.WithHTTPTrace())`
	`167`	`+ opts = append(opts, registry.WithHTTPTrace(), registry.WithClientStream(logStream))`
`167`	`168`	`}`
`168`	`169`	`reg, err := registry.NewOCIRegistry(ctx, ref, opts...)`
`169`	`170`	`if err != nil {`
Original file line number	Diff line number	Diff line change
`@@ -101,7 +101,7 @@ func GetResolver(ctx context.Context, cliContext *cli.Context) (remotes.Resolver`
`101`	`101`
`102`	`102`	`if cliContext.Bool("http-dump") {`
`103`	`103`	`hostOptions.UpdateClient = func(client *http.Client) error {`
`104`		`- httpdbg.DumpRequests(ctx, client)`
	`104`	`+ httpdbg.DumpRequests(ctx, client, nil)`
`105`	`105`	`return nil`
`106`	`106`	`}`
`107`	`107`	`}`
Original file line number	Diff line number	Diff line change
`@@ -62,10 +62,14 @@ func (t debugTransport) RoundTrip(req http.Request) (http.Response, error) {`
`62`	`62`	`}`
`63`	`63`
`64`	`64`	`// DumpRequests wraps the underlying http.Client transport to logs all requests/responses to the log.`
`65`		`-func DumpRequests(ctx context.Context, client *http.Client) {`
	`65`	`+func DumpRequests(ctx context.Context, client *http.Client, writer io.Writer) {`
	`66`	`+ if writer == nil {`
	`67`	`+ writer = log.G(ctx).Writer()`
	`68`	`+ }`
	`69`	`+`
`66`	`70`	`client.Transport = debugTransport{`
`67`	`71`	`transport: client.Transport,`
`68`		`- writer: log.G(ctx).Writer(),`
	`72`	`+ writer: writer,`
`69`	`73`	`}`
`70`	`74`	`}`
`71`	`75`