diff --git a/app/artifact-cas/README.md b/app/artifact-cas/README.md index cfc0bc8eb..18f3b7a71 100644 --- a/app/artifact-cas/README.md +++ b/app/artifact-cas/README.md @@ -67,6 +67,67 @@ go run main.go --insecure artifact upload -f [FILE] # will return a sha256 diges go run main.go --insecure artifact download -d [sha256:...] ``` + +### Shared endpoint + +The CAS HTTP port, `8001`, is the shared endpoint, mirroring the control plane (spec D-020). One standard HTTP listener owns every connection and dispatches each request: native gRPC (HTTP/2 + an `application/grpc*` content type) goes through an in-process `httputil.ReverseProxy` to a private native gRPC server bound to an ephemeral loopback address, and everything else uses the existing HTTP routes. ByteStream keeps the grpc-go native transport: the proxy streams the request body into the private connection, whose HTTP/2 flow-control window paces the proxy's reads, so no unbounded read-ahead exists in either direction. Earlier designs on this branch measured why: the per-request `grpc.Server.ServeHTTP` adapter peaked at ~1.25 GiB of live heap for a 1 GiB upload, and connection classification (cmux) could not serve the all-HTTP/2 upstream that a single-location ingress produces. Mixed native-gRPC and plain-HTTP requests share one upstream connection in any order, as on the control plane. CAS does not add gRPC-Web or Connect. + +The private native server runs the same services and interceptors as the legacy listener and is independent of it. The native gRPC listener on `9001` remains enabled by default, with its existing independent TLS configuration and flag; the shared endpoint works with it disabled. `server.http.timeout` continues to apply to HTTP routes; proxied gRPC requests bypass the HTTP route middleware. Shutdown drains the HTTP server first — including every proxied gRPC stream — then stops the private native server. + +To test with your existing development infrastructure: + +```sh +make -C app/artifact-cas run + +# In another terminal, use a per-command CAS endpoint override. +# Existing CLI authentication and the control-plane endpoint come from your config. +go run ./app/cli --insecure --artifact-cas localhost:8001 artifact upload -f /path/to/artifact +# Use the returned digest: +go run ./app/cli --insecure --artifact-cas localhost:8001 artifact download -d sha256:DIGEST +``` + +The development CAS public key and credentials service must match the control plane that issues these temporary CAS tokens. If another CAS already occupies these ports, start this binary with a copy of your CAS configuration and unused addresses. Do not replace the running service or edit the CLI config just to run a comparison. + +Change only the per-command override to `localhost:9001` to compare with the native listener. Once all direct clients and the control-plane CAS connection have been moved, `server.legacy_grpc_listener: false` disables the native listener; the shared endpoint's private native server keeps running on its loopback address either way. Leave the legacy listener enabled during evaluation. Optional `server.http.tls_config` configures TLS for the shared endpoint; both certificate and private key must be set. It does not inherit `server.grpc.tls_config`. + +The chart's `cas.ingress` mirrors the control plane: one catch-all route on one hostname, one Service port, one certificate, and an HTTP/2 (gRPC) upstream — so native gRPC, browser downloads, and status probes all arrive on the shared listener, including mixed requests on one upstream connection. The ingress keeps the legacy ingress's `100m` body limit and `3M` client buffer default. Override the body limit to `0` or a suitable larger value for a 1 GiB ingress experiment. + +`cas.ingressAPI`, `cas.serviceAPI`, and the control plane's connection to the legacy CAS Service keep their existing settings. F5 same-host routing and Traefik's per-Service scheme still need qualification. + +The HTTP server applies its 60-second header timeout; proxied gRPC streams carry no such deadline. Metrics remain plaintext on their separate listener; `server.http_metrics.tls_config` is rejected. + +Run the isolated functional checks without Vault or an object store: + +```sh +go test -race ./app/artifact-cas/internal/server ./app/artifact-cas/cmd ./deployment/chainloop -count=1 +``` + +Compare real ByteStream upload and download paths over TCP with a synthetic 1 GiB zero-filled artifact, streamed from a fixed-size buffer and checked by SHA256. Both listeners use the same file-backed test backend and real CAS staging and authentication. Select a disk-backed temporary directory; `/tmp` can be tmpfs. + +```sh +mkdir -p "$HOME/.cache/chainloop-cas-bench" +TMPDIR="$HOME/.cache/chainloop-cas-bench" go test ./app/artifact-cas/internal/server \ + -run '^$' -bench '^BenchmarkCASByteStreamTransport$' -benchtime=1x -count=3 -timeout=10m +``` + +`CAS_BENCH_BYTES` changes the artifact size for shorter experiments. Benchmark throughput and total allocated bytes come from Go's benchmark runner. `peak-heap-MiB` samples live heap every 10 ms for the combined client and server process; it is neither server-only RSS nor a bounded-memory proof. This controlled transport comparison does not establish the original R-003 requirement of throughput within 10% of the current release with representative object storage, or cover slow readers, concurrent streams, ingress, TLS throughput, or the complete R-002 authentication matrix. + + +#### Observed local results + +Three 1 GiB transfers per operation, Go 1.27.2, grpc-go 1.84.0, Linux amd64, disk-backed btrfs staging. These medians come from the [raw benchmark output](https://github.com/chainloop-dev/chainloop/pull/3663), reproduced with the command above. + +| Listener | Operation | Throughput (MB/s) | Sampled peak live heap (MiB) | +| --- | --- | ---: | ---: | +| Native gRPC (`:9001`) | Upload | 233.92 | 29.13 | +| Shared endpoint (`:8001`, proxy) | Upload | 212.89 | 32.45 | +| Native gRPC (`:9001`) | Download | 192.82 | 30.82 | +| Shared endpoint (`:8001`, proxy) | Download | 177.11 | 42.78 | + +ByteStream keeps the native transport behind the proxy: median upload throughput on the shared endpoint was 9.0% lower and median download throughput 8.1% lower than the legacy listener — within the R-003 10% shape, with a thin margin — and sampled peak live heap was ~32-43 MiB against ~29-32 MiB native in these 1 GiB runs. Four concurrent 128 MiB uploads through the dispatcher peaked at 58 MiB combined (regression-tested), so per-stream buffers do not accumulate toward the transferred bytes. The extra hop costs the double HTTP/2 framing and one copy. Filesystem caching, synthetic content, the combined client/server process, and local machine load limit extrapolation to production; the release and object-store qualification remains open. + +Earlier designs on this branch measured the alternatives. The per-request `grpc.Server.ServeHTTP` adapter peaked at ~1278 MiB of live heap on a 1 GiB upload: grpc-go's [handler transport reads the HTTP body in a separate goroutine](https://github.com/grpc/grpc-go/blob/v1.84.0/internal/transport/handler_server.go#L434) into an [unbounded receive buffer](https://github.com/grpc/grpc-go/blob/v1.84.0/internal/transport/transport.go#L90), so read-ahead is not bounded by the application's pace. Connection classification through cmux matched native throughput (upload +3.3%, download −4.3%) but could not serve the all-HTTP/2 upstream of a single-location ingress: a silent matcher deadlocks grpc-go clients, the SETTINGS-writing matcher poisons plain-HTTP/2 connections (GOAWAY, PROTOCOL_ERROR), and pooled mixed connections fail. Both results are preserved in the PR #3663 history. + ### Run tests ``` diff --git a/app/artifact-cas/cmd/main.go b/app/artifact-cas/cmd/main.go index 1e7aa4a69..c9f1ece79 100644 --- a/app/artifact-cas/cmd/main.go +++ b/app/artifact-cas/cmd/main.go @@ -37,8 +37,7 @@ import ( "github.com/go-kratos/kratos/v2/config/env" "github.com/go-kratos/kratos/v2/config/file" "github.com/go-kratos/kratos/v2/log" - "github.com/go-kratos/kratos/v2/transport/grpc" - "github.com/go-kratos/kratos/v2/transport/http" + "github.com/go-kratos/kratos/v2/transport" "go.opentelemetry.io/otel/trace" _ "go.uber.org/automaxprocs" @@ -66,7 +65,11 @@ type app struct { backend.Providers } -func newApp(logger log.Logger, gs *grpc.Server, hs *http.Server, ms *server.HTTPMetricsServer, providers backend.Providers, _ trace.TracerProvider) *app { +func newApp(logger log.Logger, gs *server.GRPCServers, hs *server.APIServer, ms *server.HTTPMetricsServer, providers backend.Providers, _ trace.TracerProvider) *app { + servers := []transport.Server{hs, ms} + if gs.Legacy != nil { + servers = append(servers, gs.Legacy) + } return &app{ kratos.New( kratos.ID(id), @@ -74,11 +77,7 @@ func newApp(logger log.Logger, gs *grpc.Server, hs *http.Server, ms *server.HTTP kratos.Version(Version), kratos.Metadata(map[string]string{}), kratos.Logger(logger), - kratos.Server( - gs, - hs, - ms, - ), + kratos.Server(servers...), ), providers, } diff --git a/app/artifact-cas/cmd/wire_gen.go b/app/artifact-cas/cmd/wire_gen.go index aa28bcf9a..5553b6f05 100644 --- a/app/artifact-cas/cmd/wire_gen.go +++ b/app/artifact-cas/cmd/wire_gen.go @@ -48,26 +48,36 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth v := serviceOpts(logger, auditDispatcher, bootstrap, cache) byteStreamService := service.NewByteStreamService(providers, v...) resourceService := service.NewResourceService(providers, v...) + downloadService := service.NewDownloadService(providers, v...) validator, err := newProtoValidator() if err != nil { cleanup2() cleanup() return nil, nil, err } - grpcServer, err := server.NewGRPCServer(confServer, auth, byteStreamService, resourceService, providers, validator, logger) + opts := &server.Opts{ + ServerConfig: confServer, + AuthConfig: auth, + ByteService: byteStreamService, + ResourceSvc: resourceService, + DownloadSvc: downloadService, + Providers: providers, + Validator: validator, + Logger: logger, + } + grpcServers, err := server.NewGRPCServers(opts) if err != nil { cleanup2() cleanup() return nil, nil, err } - downloadService := service.NewDownloadService(providers, v...) - httpServer, err := server.NewHTTPServer(confServer, auth, downloadService, providers, logger) + apiServer, err := server.NewHTTPServer(opts, grpcServers) if err != nil { cleanup2() cleanup() return nil, nil, err } - httpMetricsServer, err := server.NewHTTPMetricsServer(confServer) + httpMetricsServer, err := server.NewHTTPMetricsServer(confServer, logger) if err != nil { cleanup2() cleanup() @@ -79,7 +89,7 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth cleanup() return nil, nil, err } - mainApp := newApp(logger, grpcServer, httpServer, httpMetricsServer, providers, tracerProvider) + mainApp := newApp(logger, grpcServers, apiServer, httpMetricsServer, providers, tracerProvider) return mainApp, func() { cleanup3() cleanup2() diff --git a/app/artifact-cas/configs/config.devel.yaml b/app/artifact-cas/configs/config.devel.yaml index 65936c30c..83863e30d 100644 --- a/app/artifact-cas/configs/config.devel.yaml +++ b/app/artifact-cas/configs/config.devel.yaml @@ -3,7 +3,12 @@ # Please make sure you run those via the convenience docker compose file # [root]/devel/compose.yml server: + # Keep 9001 available to compare native gRPC with the shared 8001 endpoint. + legacy_grpc_listener: true http: + # Shared endpoint: HTTP/1.1, HTTP/2, and native gRPC. Each request is + # dispatched; native gRPC reaches the native transport through an + # in-process reverse proxy (spec D-020). addr: 0.0.0.0:8001 # Timeouts for http downloads # grpc downloads/uploads don't require this because they don't have timeouts @@ -24,7 +29,7 @@ server: credentials_service: # we will check that we can read there - secret_prefix: chainloop-devel + secret_prefix: chainloop-devel # checkov:skip=CKV_SECRET_6: development-only credential-path prefix, not a credential # Remember to run vault via docker compose up vault: address: ${VAULT_ADDRESS:http://0.0.0.0:8200} diff --git a/app/artifact-cas/configs/samples/config.yaml b/app/artifact-cas/configs/samples/config.yaml index be3e16fd0..7807a93fc 100644 --- a/app/artifact-cas/configs/samples/config.yaml +++ b/app/artifact-cas/configs/samples/config.yaml @@ -3,9 +3,18 @@ auth: public_key_path: "./configs/devkeys/cas.public.pem" server: + # Retain the native transport while evaluating the shared HTTP endpoint. + legacy_grpc_listener: true http: addr: 0.0.0.0:8001 - timeout: 1s + timeout: 300s + # Shared endpoint: HTTP/1.1, HTTP/2, and native gRPC, dispatched per + # request; native gRPC reaches the native transport through an in-process + # reverse proxy (spec D-020). + # Optional TLS for both transports on port 8001. Independent of grpc.tls_config. + # tls_config: + # certificate: "./configs/tls/shared.crt" + # private_key: "./configs/tls/shared.key" grpc: addr: 0.0.0.0:9001 timeout: 1s @@ -17,7 +26,7 @@ server: credentials_service: # We use the prefix to check that we can read from it on initialization - secret_prefix: chainloop-devel + secret_prefix: chainloop-devel # checkov:skip=CKV_SECRET_6: development-only credential-path prefix, not a credential # Remember to run vault via docker compose up vault: address: ${VAULT_ADDRESS:http://0.0.0.0:8200} @@ -26,7 +35,7 @@ credentials_service: # aws_secret_manager: # creds: # access_key: not-a-key - # secret_key: not-a-secret + # secret_key: not-a-secret # checkov:skip=CKV_SECRET_6: commented-out example value, not a credential # region: us-east-1 # gcp_secret_manager: diff --git a/app/artifact-cas/internal/conf/conf.pb.go b/app/artifact-cas/internal/conf/conf.pb.go index ac72ca7ae..8e33715e5 100644 --- a/app/artifact-cas/internal/conf/conf.pb.go +++ b/app/artifact-cas/internal/conf/conf.pb.go @@ -131,14 +131,16 @@ func (x *Bootstrap) GetStagingDir() string { type Server struct { state protoimpl.MessageState `protogen:"open.v1"` - // Regular HTTP endpoint + // Shared HTTP and native gRPC endpoint Http *Server_HTTP `protobuf:"bytes,1,opt,name=http,proto3" json:"http,omitempty"` // GRPC API endpoint Grpc *Server_GRPC `protobuf:"bytes,2,opt,name=grpc,proto3" json:"grpc,omitempty"` // hHTTP server where the prometheus metrics will get exposed - HttpMetrics *Server_HTTP `protobuf:"bytes,3,opt,name=http_metrics,json=httpMetrics,proto3" json:"http_metrics,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + HttpMetrics *Server_HTTP `protobuf:"bytes,3,opt,name=http_metrics,json=httpMetrics,proto3" json:"http_metrics,omitempty"` + // The native gRPC listener remains enabled when this field is unset. + LegacyGrpcListener *bool `protobuf:"varint,4,opt,name=legacy_grpc_listener,json=legacyGrpcListener,proto3,oneof" json:"legacy_grpc_listener,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Server) Reset() { @@ -192,6 +194,13 @@ func (x *Server) GetHttpMetrics() *Server_HTTP { return nil } +func (x *Server) GetLegacyGrpcListener() bool { + if x != nil && x.LegacyGrpcListener != nil { + return *x.LegacyGrpcListener + } + return false +} + type Auth struct { state protoimpl.MessageState `protogen:"open.v1"` // Public key used to verify the received JWT token @@ -553,7 +562,9 @@ type Server_HTTP struct { Addr string `protobuf:"bytes,2,opt,name=addr,proto3" json:"addr,omitempty"` Timeout *durationpb.Duration `protobuf:"bytes,3,opt,name=timeout,proto3" json:"timeout,omitempty"` // CORS configuration for the HTTP download endpoint - Cors *Server_CORS `protobuf:"bytes,4,opt,name=cors,proto3" json:"cors,omitempty"` + Cors *Server_CORS `protobuf:"bytes,4,opt,name=cors,proto3" json:"cors,omitempty"` + // TLS for every protocol on the shared HTTP endpoint. + TlsConfig *Server_TLS `protobuf:"bytes,5,opt,name=tls_config,json=tlsConfig,proto3" json:"tls_config,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -616,6 +627,13 @@ func (x *Server_HTTP) GetCors() *Server_CORS { return nil } +func (x *Server_HTTP) GetTlsConfig() *Server_TLS { + if x != nil { + return x.TlsConfig + } + return nil +} + type Server_TLS struct { state protoimpl.MessageState `protogen:"open.v1"` // path to certificate and private key @@ -768,18 +786,21 @@ const file_conf_proto_rawDesc = "" + "\bendpoint\x18\x02 \x01(\tR\bendpoint\x12\x1a\n" + "\binsecure\x18\x03 \x01(\bR\binsecure\x12*\n" + "\x0esampling_ratio\x18\x04 \x01(\x01H\x00R\rsamplingRatio\x88\x01\x01B\x11\n" + - "\x0f_sampling_ratio\"\x9a\x04\n" + + "\x0f_sampling_ratio\"\x96\x05\n" + "\x06Server\x12 \n" + "\x04http\x18\x01 \x01(\v2\f.Server.HTTPR\x04http\x12 \n" + "\x04grpc\x18\x02 \x01(\v2\f.Server.GRPCR\x04grpc\x12/\n" + - "\fhttp_metrics\x18\x03 \x01(\v2\f.Server.HTTPR\vhttpMetrics\x1a+\n" + + "\fhttp_metrics\x18\x03 \x01(\v2\f.Server.HTTPR\vhttpMetrics\x125\n" + + "\x14legacy_grpc_listener\x18\x04 \x01(\bH\x00R\x12legacyGrpcListener\x88\x01\x01\x1a+\n" + "\x04CORS\x12#\n" + - "\rallow_origins\x18\x01 \x03(\tR\fallowOrigins\x1a\x8b\x01\n" + + "\rallow_origins\x18\x01 \x03(\tR\fallowOrigins\x1a\xb7\x01\n" + "\x04HTTP\x12\x18\n" + "\anetwork\x18\x01 \x01(\tR\anetwork\x12\x12\n" + "\x04addr\x18\x02 \x01(\tR\x04addr\x123\n" + "\atimeout\x18\x03 \x01(\v2\x19.google.protobuf.DurationR\atimeout\x12 \n" + - "\x04cors\x18\x04 \x01(\v2\f.Server.CORSR\x04cors\x1aH\n" + + "\x04cors\x18\x04 \x01(\v2\f.Server.CORSR\x04cors\x12*\n" + + "\n" + + "tls_config\x18\x05 \x01(\v2\v.Server.TLSR\ttlsConfig\x1aH\n" + "\x03TLS\x12 \n" + "\vcertificate\x18\x01 \x01(\tR\vcertificate\x12\x1f\n" + "\vprivate_key\x18\x02 \x01(\tR\n" + @@ -789,7 +810,8 @@ const file_conf_proto_rawDesc = "" + "\x04addr\x18\x02 \x01(\tR\x04addr\x123\n" + "\atimeout\x18\x03 \x01(\v2\x19.google.protobuf.DurationR\atimeout\x12*\n" + "\n" + - "tls_config\x18\x04 \x01(\v2\v.Server.TLSR\ttlsConfig\"t\n" + + "tls_config\x18\x04 \x01(\v2\v.Server.TLSR\ttlsConfigB\x17\n" + + "\x15_legacy_grpc_listener\"t\n" + "\x04Auth\x12D\n" + "\x1drobot_account_public_key_path\x18\x01 \x01(\tB\x02\x18\x01R\x19robotAccountPublicKeyPath\x12&\n" + "\x0fpublic_key_path\x18\x02 \x01(\tR\rpublicKeyPathBHZFgithub.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf;confb\x06proto3" @@ -835,13 +857,14 @@ var file_conf_proto_depIdxs = []int32{ 6, // 9: Bootstrap.Observability.tracing:type_name -> Bootstrap.Observability.Tracing 12, // 10: Server.HTTP.timeout:type_name -> google.protobuf.Duration 7, // 11: Server.HTTP.cors:type_name -> Server.CORS - 12, // 12: Server.GRPC.timeout:type_name -> google.protobuf.Duration - 9, // 13: Server.GRPC.tls_config:type_name -> Server.TLS - 14, // [14:14] is the sub-list for method output_type - 14, // [14:14] is the sub-list for method input_type - 14, // [14:14] is the sub-list for extension type_name - 14, // [14:14] is the sub-list for extension extendee - 0, // [0:14] is the sub-list for field type_name + 9, // 12: Server.HTTP.tls_config:type_name -> Server.TLS + 12, // 13: Server.GRPC.timeout:type_name -> google.protobuf.Duration + 9, // 14: Server.GRPC.tls_config:type_name -> Server.TLS + 15, // [15:15] is the sub-list for method output_type + 15, // [15:15] is the sub-list for method input_type + 15, // [15:15] is the sub-list for extension type_name + 15, // [15:15] is the sub-list for extension extendee + 0, // [0:15] is the sub-list for field type_name } func init() { file_conf_proto_init() } @@ -849,6 +872,7 @@ func file_conf_proto_init() { if File_conf_proto != nil { return } + file_conf_proto_msgTypes[1].OneofWrappers = []any{} file_conf_proto_msgTypes[3].OneofWrappers = []any{ (*Bootstrap_NatsServer_Token)(nil), } diff --git a/app/artifact-cas/internal/conf/conf.proto b/app/artifact-cas/internal/conf/conf.proto index d9c8c5399..bf7ed41eb 100644 --- a/app/artifact-cas/internal/conf/conf.proto +++ b/app/artifact-cas/internal/conf/conf.proto @@ -76,6 +76,8 @@ message Server { google.protobuf.Duration timeout = 3; // CORS configuration for the HTTP download endpoint CORS cors = 4; + // TLS for every protocol on the shared HTTP endpoint. + TLS tls_config = 5; } message TLS { // path to certificate and private key @@ -88,12 +90,14 @@ message Server { google.protobuf.Duration timeout = 3; TLS tls_config = 4; } - // Regular HTTP endpoint + // Shared HTTP and native gRPC endpoint HTTP http = 1; // GRPC API endpoint GRPC grpc = 2; // hHTTP server where the prometheus metrics will get exposed HTTP http_metrics = 3; + // The native gRPC listener remains enabled when this field is unset. + optional bool legacy_grpc_listener = 4; } message Auth { diff --git a/app/artifact-cas/internal/server/grpc.go b/app/artifact-cas/internal/server/grpc.go index f497a613f..3bddc9e8a 100644 --- a/app/artifact-cas/internal/server/grpc.go +++ b/app/artifact-cas/internal/server/grpc.go @@ -20,6 +20,7 @@ import ( "crypto/ecdsa" "crypto/tls" "fmt" + "net" "os" "regexp" @@ -51,8 +52,107 @@ import ( grpcLib "google.golang.org/grpc" ) -// NewGRPCServer new a gRPC server. -func NewGRPCServer(c *conf.Server, authConf *conf.Auth, byteService *service.ByteStreamService, rSvc *service.ResourceService, providers backend.Providers, validator protovalidate.Validator, logger log.Logger) (*grpc.Server, error) { +// Opts provides the shared service implementations and transport settings. +type Opts struct { + ServerConfig *conf.Server + AuthConfig *conf.Auth + ByteService *service.ByteStreamService + ResourceSvc *service.ResourceService + DownloadSvc *service.DownloadService + Providers backend.Providers + Validator protovalidate.Validator + Logger log.Logger +} + +// GRPCServers keeps the shared endpoint's private native gRPC server separate +// from the optional legacy listener. Both run the same services and +// interceptors. Private binds an ephemeral loopback listener that the shared +// endpoint's reverse proxy dials; it is independent of the legacy listener's +// address, TLS configuration, and flag, and exists whether or not the legacy +// listener is enabled. +type GRPCServers struct { + Private *grpc.Server + PrivateAddr string + // PrivateLis holds the private listener so a caller that never starts + // the server can still close the socket: before Start registers the + // listener with grpc-go, Private.Stop does not close it. + PrivateLis net.Listener + Legacy *grpc.Server +} + +// Close releases the private socket without serving. Callers that hand the +// servers to the application lifecycle must not call it; construction-failure +// paths use it to avoid leaking the bound address. +func (s *GRPCServers) Close() error { + if s.PrivateLis == nil { + return nil + } + err := s.PrivateLis.Close() + s.PrivateLis = nil + return err +} + +func legacyGRPCListenerEnabled(c *conf.Server) bool { + return c.LegacyGrpcListener == nil || *c.LegacyGrpcListener +} + +// NewGRPCServers builds the private native server for the shared endpoint and +// the optional legacy listener, with the same services and interceptors. +func NewGRPCServers(opts *Opts) (*GRPCServers, error) { + // Bind the private listener here so the shared endpoint's proxy knows its + // address at construction. Loopback-only, ephemeral: never public, never + // configurable, and independent of server.grpc.addr (which may carry TLS). + privateLis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return nil, fmt.Errorf("binding the shared endpoint's private gRPC listener: %w", err) + } + private, err := newGRPCServer(opts, grpc.Listener(privateLis)) + if err != nil { + _ = privateLis.Close() + return nil, err + } + servers := &GRPCServers{Private: private, PrivateAddr: privateLis.Addr().String(), PrivateLis: privateLis} + if legacyGRPCListenerEnabled(opts.ServerConfig) { + servers.Legacy, err = NewGRPCServer(opts) + if err != nil { + private.Server.Stop() + _ = servers.Close() + return nil, err + } + } + return servers, nil +} + +// NewGRPCServer retains the existing native listener and its independent TLS. +func NewGRPCServer(opts *Opts) (*grpc.Server, error) { + c := opts.ServerConfig.GetGrpc() + var serverOpts []grpc.ServerOption + if c.GetNetwork() != "" { + serverOpts = append(serverOpts, grpc.Network(c.GetNetwork())) + } + if c.GetAddr() != "" { + serverOpts = append(serverOpts, grpc.Address(c.GetAddr())) + } + if tlsConf := c.GetTlsConfig(); tlsConf != nil { + cert, key := tlsConf.GetCertificate(), tlsConf.GetPrivateKey() + // Keep the legacy behavior for partial TLS configuration. + if cert != "" && key != "" { + pair, err := tls.LoadX509KeyPair(cert, key) + if err != nil { + return nil, fmt.Errorf("loading gRPC server TLS certificate: %w", err) + } + serverOpts = append(serverOpts, grpc.TLSConfig(&tls.Config{ + Certificates: []tls.Certificate{pair}, MinVersion: tls.VersionTLS12, + })) + } + } + return newGRPCServer(opts, serverOpts...) +} + +func newGRPCServer(settings *Opts, extraOpts ...grpc.ServerOption) (*grpc.Server, error) { + c, authConf := settings.ServerConfig, settings.AuthConfig + byteService, rSvc := settings.ByteService, settings.ResourceSvc + providers, validator, logger := settings.Providers, settings.Validator, settings.Logger // Parse the public key once on initialization instead of on every request // TODO: implement jwks endpoint publicKey, err := parsePublicKey(authConf, logger) @@ -104,28 +204,9 @@ func NewGRPCServer(c *conf.Server, authConf *conf.Auth, byteService *service.Byt // Since we track uploads / downloads we'll increase the buckets grpc_prometheus.EnableHandlingTimeHistogram(grpc_prometheus.WithHistogramBuckets(prometheus.ExponentialBucketsRange(0.5, 60, 8))) - if c.Grpc.Network != "" { - opts = append(opts, grpc.Network(c.Grpc.Network)) - } - if c.Grpc.Addr != "" { - opts = append(opts, grpc.Address(c.Grpc.Addr)) - } - if c.Grpc.Timeout != nil { - opts = append(opts, grpc.Timeout(c.Grpc.Timeout.AsDuration())) - } - if tlsConf := c.Grpc.GetTlsConfig(); tlsConf != nil { - cert := tlsConf.GetCertificate() - privKey := tlsConf.GetPrivateKey() - if cert != "" && privKey != "" { - cert, err := tls.LoadX509KeyPair(cert, privKey) - if err != nil { - return nil, fmt.Errorf("loading gRPC server TLS certificate: %w", err) - } - opts = append(opts, grpc.TLSConfig(&tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS12, // gosec complains about insecure minimum version we use default value - })) - } + opts = append(opts, extraOpts...) + if c.GetGrpc().GetTimeout() != nil { + opts = append(opts, grpc.Timeout(c.GetGrpc().GetTimeout().AsDuration())) } srv := grpc.NewServer(opts...) diff --git a/app/artifact-cas/internal/server/http.go b/app/artifact-cas/internal/server/http.go index d38873340..1c48c2aec 100644 --- a/app/artifact-cas/internal/server/http.go +++ b/app/artifact-cas/internal/server/http.go @@ -16,50 +16,265 @@ package server import ( + "context" + "crypto/tls" + "errors" + "fmt" + "net" + h "net/http" + "net/http/httputil" + "net/url" + "strings" + "sync" + "sync/atomic" + "time" + api "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/service" casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" - backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" middlewares_http "github.com/chainloop-dev/chainloop/pkg/middlewares/http" "github.com/go-kratos/kratos/v2/log" "github.com/go-kratos/kratos/v2/middleware/logging" "github.com/go-kratos/kratos/v2/middleware/recovery" + "github.com/go-kratos/kratos/v2/transport/grpc" "github.com/go-kratos/kratos/v2/transport/http" "github.com/golang-jwt/jwt/v5" + grpcLib "google.golang.org/grpc" ) -// NewHTTPServer new a HTTP server. -func NewHTTPServer(c *conf.Server, authConf *conf.Auth, downloadSvc *service.DownloadService, providers backend.Providers, logger log.Logger) (*http.Server, error) { +// APIServer owns the shared CAS listener on server.http.addr. The standard +// HTTP server owns every connection and dispatches each request: +// +// native gRPC (HTTP/2 + application/grpc*) -> httputil.ReverseProxy -> +// the private native gRPC server (grpc.Server.Serve, native transport) +// everything else -> the existing Kratos HTTP routes +// +// ByteStream keeps the native transport: the proxy streams the request body +// into the private connection, whose HTTP/2 flow-control window paces the +// proxy's reads, and net/http's own window paces the client. No unbounded +// read-ahead exists anywhere on the path, in either direction. Mixed gRPC and +// ordinary HTTP requests may share one upstream connection in any order, as +// they do on the control plane (spec D-020). +type APIServer struct { + lis net.Listener + httpSrv *http.Server + grpcSrv *grpc.Server + scheme string + workers sync.WaitGroup + stopping atomic.Bool +} + +// Start serves the shared endpoint until ctx is cancelled or a transport +// fails. The HTTP server serves every connection; the private native gRPC +// server serves the connections the proxy opens to it. A genuine failure of +// either transport — including the private gRPC listener failing — is +// surfaced so the whole endpoint stops, rather than serving HTTP while +// proxied gRPC answers 502. +func (s *APIServer) Start(ctx context.Context) error { + // Buffered to the worker count: both sends complete into the buffer even + // when Start returns after the first error, so the second transport's + // report never blocks and the goroutine always finishes. + errCh := make(chan error, 2) + s.workers.Go(func() { errCh <- s.httpSrv.Start(ctx) }) + s.workers.Go(func() { errCh <- s.grpcSrv.Start(ctx) }) + select { + case <-ctx.Done(): + return nil + case err := <-errCh: + // The ordinary-stop sentinel (ErrClosed / ErrServerStopped) is benign + // only when it comes from our own Stop. The same error from a + // spontaneously failed transport must propagate, or the app would keep + // serving HTTP after the private gRPC server is gone. + if s.stopping.Load() && isOrdinaryStopError(err) { + return nil + } + return err + } +} + +// Stop drains the HTTP server first — which includes every proxied gRPC +// stream, because those live on HTTP connections — then stops the private +// native gRPC server so its in-flight RPCs finish. +func (s *APIServer) Stop(ctx context.Context) error { + s.stopping.Store(true) + httpErr := s.httpSrv.Stop(ctx) + grpcErr := s.grpcSrv.Stop(ctx) + s.workers.Wait() + return errors.Join(httpErr, grpcErr) +} + +// Endpoint returns the real bound address of the shared endpoint. +func (s *APIServer) Endpoint() (*url.URL, error) { + return &url.URL{Scheme: s.scheme, Host: s.lis.Addr().String()}, nil +} + +// NewHTTPServer binds the shared CAS endpoint and dispatches every request. +// When it returns an error, it has closed the private gRPC socket it would +// have consumed, so a failed construction does not leak it. +func NewHTTPServer(settings *Opts, grpcServers *GRPCServers) (*APIServer, error) { + c, authConf := settings.ServerConfig, settings.AuthConfig + downloadSvc, providers, logger := settings.DownloadSvc, settings.Providers, settings.Logger + if c.GetHttp().GetAddr() == "" { + _ = grpcServers.Close() + return nil, errors.New("server.http.addr must be set for the shared CAS endpoint") + } + tlsConfig, err := loadServerTLSConfig(c.GetHttp().GetTlsConfig()) + if err != nil { + _ = grpcServers.Close() + return nil, err + } + + // Validate before binding, so initialization errors do not leak a listener. + publicKey, err := parsePublicKey(authConf, logger) + if err != nil { + _ = grpcServers.Close() + return nil, err + } + + network := c.Http.GetNetwork() + if network == "" { + network = "tcp" + } + lis, err := net.Listen(network, c.Http.GetAddr()) + if err != nil { + _ = grpcServers.Close() + return nil, fmt.Errorf("binding the shared CAS endpoint %q: %w", c.Http.GetAddr(), err) + } + scheme := "http" + if tlsConfig != nil { + // One certificate covers both transports, with ALPN offering + // HTTP/2 and HTTP/1.1. + lis = tls.NewListener(lis, tlsConfig) + scheme = "https" + } + var opts = []http.ServerOption{ + // The HTTP server never binds its own socket: it serves the shared + // listener this function bound. + http.Listener(lis), http.Middleware( recovery.Recovery(), logging.Server(logger), ), } - if c.Http.Network != "" { - opts = append(opts, http.Network(c.Http.Network)) - } - if c.Http.Addr != "" { - opts = append(opts, http.Address(c.Http.Addr)) - } if c.Http.Timeout != nil { opts = append(opts, http.Timeout(c.Http.Timeout.AsDuration())) } - // Parse the public key once on initialization instead of on every request - // TODO: implement jwks endpoint - publicKey, err := parsePublicKey(authConf, logger) - if err != nil { - return nil, err - } - srv := http.NewServer(opts...) downloadHandler := middlewares_http.AuthFromHeaderOrQueryParam(loadPublicKey(publicKey), claimsFunc(), casJWT.SigningMethod, downloadSvc) srv.Handle(service.DownloadPath, CORSMiddleware(c.GetHttp().GetCors().GetAllowOrigins(), downloadHandler)) api.RegisterStatusServiceHTTPServer(srv, service.NewStatusService(Version, providers)) - return srv, nil + + routes := srv.Handler + srv.Protocols = new(h.Protocols) + srv.Protocols.SetHTTP1(true) + srv.Protocols.SetHTTP2(true) + srv.Protocols.SetUnencryptedHTTP2(true) + // Bound how long a client may take to deliver its request headers + // (slowloris protection). Response streaming is not affected. + srv.ReadHeaderTimeout = 60 * time.Second + // One proxy and one transport for the process: its connection pool is + // reused across requests. Dispatch each request: native gRPC goes through + // the standard reverse proxy to the private native server; everything else + // uses the existing HTTP routes. CAS has no gRPC-Web endpoint (spec + // D-011); the grpc-web exclusion keeps the content-type check consistent + // with the control plane. + proxy := grpcProxy(grpcServers.PrivateAddr, logger) + srv.Handler = h.HandlerFunc(func(w h.ResponseWriter, r *h.Request) { + // Native gRPC requires HTTP/2: an HTTP/1.1 request with an + // application/grpc content type cannot reach the private gRPC server, + // so only HTTP/2 requests take the proxy branch. + if r.ProtoMajor == 2 && isNativeGRPCContentType(r.Header.Get("Content-Type")) { + proxy.ServeHTTP(w, r) + return + } + routes.ServeHTTP(w, r) + }) + + log.NewHelper(logger).Infof("[shared] CAS endpoint (%s) listening on: %s", scheme, lis.Addr().String()) + if legacyGRPCListenerEnabled(c) { + log.NewHelper(logger).Warnw("msg", fmt.Sprintf("the gRPC-only listener on %q is deprecated; use the shared HTTP endpoint and set server.legacy_grpc_listener to false when all clients have moved", c.GetGrpc().GetAddr())) + } + + return &APIServer{ + lis: lis, + httpSrv: srv, + grpcSrv: grpcServers.Private, + scheme: scheme, + }, nil +} + +// grpcProxy builds the reverse proxy that carries native gRPC requests to the +// private native gRPC server over h2c. httputil.ReverseProxy streams request +// and response bodies, forwards hop-by-hop-excluded headers (including gRPC +// metadata such as the authorization token), and copies response trailers, +// so the existing gRPC frames pass through without decoding or rebuilding. +// gRPC responses carry no Content-Length, so ReverseProxy flushes each write +// immediately through its streaming special case, which gRPC streaming +// requires. The transport speaks only unencrypted HTTP/2 (h2c prior +// knowledge): the private server is gRPC-only and loopback-only. The +// Rewrite restores the incoming authority after SetURL clears it, so the +// native server receives the original :authority for interceptor parity. +// +// The upstream pool targets a single loopback host and HTTP/2 multiplexes +// many streams onto one connection, so a small connection cap is plenty; the +// cap exists to bound idle loopback FDs held between requests, not to shape +// throughput. Idle connections are reaped after the 90-second idle timeout so +// a drained transfer does not pin the pooled connection for the process +// lifetime; Stop drains HTTP first, so idle upstream sockets only persist +// briefly past the last request. ExpectContinueTimeout is a no-op for gRPC +// (clients never send Expect: 100-continue) and does not hold a pool slot +// beyond the dial. +func grpcProxy(target string, logger log.Logger) *httputil.ReverseProxy { + // Fixed for the process lifetime: built once rather than per request. + targetURL := &url.URL{Scheme: "http", Host: target} + transport := &h.Transport{ + Protocols: new(h.Protocols), + DialContext: (&net.Dialer{Timeout: 10 * time.Second}).DialContext, + MaxIdleConns: 10, + MaxIdleConnsPerHost: 10, + IdleConnTimeout: 90 * time.Second, + ExpectContinueTimeout: time.Second, + // Byte-exact pass-through: gRPC carries its own per-message content + // coding, and the transport must neither add transport-level gzip + // headers nor transparently decode the proxied body. + DisableCompression: true, + } + transport.Protocols.SetUnencryptedHTTP2(true) + transport.Protocols.SetHTTP1(false) + return &httputil.ReverseProxy{ + Rewrite: func(pr *httputil.ProxyRequest) { + // SetURL only reads the target, so one shared URL is safe across + // concurrent requests. + pr.SetURL(targetURL) + // SetURL clears Out.Host for security (RFC 7230 host injection), + // but gRPC carries the routing target in :authority. Restore the + // incoming authority so the native server's interceptors observe + // the same value they would on a direct connection. + pr.Out.Host = pr.In.Host + }, + Transport: transport, + // On an upstream failure the handler must only write the status; when + // it returns, net/http finalizes the HTTP/2 response with END_STREAM, + // so the client gets a clean 502 instead of a stream left open. The + // regression test TestCASProxyErrorTerminatesHTTP2Stream covers this. + ErrorHandler: func(w h.ResponseWriter, _ *h.Request, err error) { + log.NewHelper(logger).Errorf("shared endpoint gRPC proxy error: %v", err) + w.WriteHeader(h.StatusBadGateway) + }, + } +} + +// isNativeGRPCContentType reports whether a content type selects the native +// gRPC protocol family, excluding gRPC-Web. CAS has no gRPC-Web endpoint, but +// the exclusion keeps the dispatch consistent with the control plane. +func isNativeGRPCContentType(contentType string) bool { + contentType = strings.ToLower(contentType) + return strings.HasPrefix(contentType, "application/grpc") && + !strings.HasPrefix(contentType, "application/grpc-web") } // claimsFunc returns the claims function for the JWT middleware that casts the claims to the correct type @@ -68,3 +283,33 @@ func claimsFunc() middlewares_http.ClaimsFunc { return &casJWT.Claims{} } } + +// isOrdinaryStopError reports whether a transport error is the expected +// result of an orderly Stop rather than a serving failure. +func isOrdinaryStopError(err error) bool { + if err == nil { + return true + } + return errors.Is(err, net.ErrClosed) || + errors.Is(err, h.ErrServerClosed) || + errors.Is(err, grpcLib.ErrServerStopped) +} + +// loadServerTLSConfig applies TLS independently of the legacy gRPC listener. +func loadServerTLSConfig(c *conf.Server_TLS) (*tls.Config, error) { + cert, key := c.GetCertificate(), c.GetPrivateKey() + if cert == "" && key == "" { + return nil, nil + } + if cert == "" || key == "" { + return nil, errors.New("incomplete TLS configuration: certificate and private key must be set together") + } + pair, err := tls.LoadX509KeyPair(cert, key) + if err != nil { + return nil, fmt.Errorf("loading shared CAS TLS certificate: %w", err) + } + return &tls.Config{ + Certificates: []tls.Certificate{pair}, MinVersion: tls.VersionTLS12, + NextProtos: []string{"h2", "http/1.1"}, + }, nil +} diff --git a/app/artifact-cas/internal/server/http_test.go b/app/artifact-cas/internal/server/http_test.go new file mode 100644 index 000000000..c99ed550b --- /dev/null +++ b/app/artifact-cas/internal/server/http_test.go @@ -0,0 +1,981 @@ +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "bytes" + "context" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/base64" + "encoding/gob" + "encoding/hex" + "encoding/pem" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "buf.build/go/protovalidate" + v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" + "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" + "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/service" + casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" + backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" + "github.com/chainloop-dev/chainloop/pkg/blobmanager/mocks" + "github.com/go-kratos/kratos/v2/log" + "github.com/go-kratos/kratos/v2/transport" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "google.golang.org/genproto/googleapis/bytestream" + grpcLib "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + reflection "google.golang.org/grpc/reflection/grpc_reflection_v1" + "google.golang.org/grpc/status" +) + +const ( + casAnyAddr = "127.0.0.1:0" + casTestFileName = "artifact.bin" + casSharedEndpoint = "shared" + casLegacyEndpoint = "legacy" + casGRPCRequest = "grpc" + casHTTPRequest = "http" + casPreflightRequest = "preflight" + casExpiredTokenCase = "expired token" +) + +// diskBackend stores the bytes rather than mocking the service's upload and +// download results. It keeps test artifact contents off the Go heap. +type diskBackend struct{ dir string } + +func (b *diskBackend) Upload(_ context.Context, r io.Reader, resource *v1.CASResource) error { + f, err := os.Create(filepath.Join(b.dir, resource.Digest)) + if err != nil { + return err + } + _, copyErr := io.Copy(f, r) + closeErr := f.Close() + if copyErr != nil { + return copyErr + } + return closeErr +} +func (b *diskBackend) Download(_ context.Context, w io.Writer, digest string) error { + f, err := os.Open(filepath.Join(b.dir, digest)) + if err != nil { + return err + } + defer f.Close() + _, err = io.Copy(w, f) + return err +} +func (b *diskBackend) Describe(_ context.Context, digest string) (*v1.CASResource, error) { + info, err := os.Stat(filepath.Join(b.dir, digest)) + if err != nil { + return nil, err + } + return &v1.CASResource{Digest: digest, FileName: casTestFileName, Size: info.Size()}, nil +} +func (b *diskBackend) Exists(_ context.Context, digest string) (bool, error) { + _, err := os.Stat(filepath.Join(b.dir, digest)) + if os.IsNotExist(err) { + return false, nil + } + return err == nil, err +} +func (*diskBackend) CheckWritePermissions(context.Context) error { return nil } + +func casTestOpts(t testing.TB) *Opts { + t.Helper() + storage := &diskBackend{dir: t.TempDir()} + provider := mocks.NewProvider(t) + provider.On("FromCredentials", mock.Anything, "test-secret").Maybe().Return(storage, nil) + providers := backend.Providers{"test": provider} + validator, err := protovalidate.New() + require.NoError(t, err) + svcOpts := []service.NewOpt{service.WithStagingDir(t.TempDir())} + return &Opts{ + ServerConfig: &conf.Server{ + Http: &conf.Server_HTTP{Addr: casAnyAddr, Cors: &conf.Server_CORS{AllowOrigins: []string{"https://example.test"}}}, + Grpc: &conf.Server_GRPC{Addr: casAnyAddr}, + }, + AuthConfig: &conf.Auth{PublicKeyPath: "testdata/test-key.ec.pub"}, + ByteService: service.NewByteStreamService(providers, svcOpts...), + ResourceSvc: service.NewResourceService(providers, svcOpts...), + DownloadSvc: service.NewDownloadService(providers, svcOpts...), + Providers: providers, Validator: validator, Logger: log.NewStdLogger(io.Discard), + } +} + +func startCASTestServers(t testing.TB, opts *Opts) (sharedURL, legacyAddr string) { + t.Helper() + servers, err := NewGRPCServers(opts) + require.NoError(t, err) + api, err := NewHTTPServer(opts, servers) + require.NoError(t, err) + endpoint, err := api.Endpoint() + require.NoError(t, err) + startCASTestTransport(t, api) + if servers.Legacy != nil { + legacy, err := servers.Legacy.Endpoint() + require.NoError(t, err) + legacyAddr = legacy.Host + startCASTestTransport(t, servers.Legacy) + } + return endpoint.String(), legacyAddr +} + +func startCASTestTransport(t testing.TB, srv transport.Server) { + t.Helper() + done := make(chan error, 1) + go func() { done <- srv.Start(context.Background()) }() + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + require.NoError(t, srv.Stop(ctx)) + select { + case err := <-done: + require.NoError(t, err) + case <-ctx.Done(): + require.FailNow(t, "CAS server did not stop") + } + }) +} + +func casTestConn(t testing.TB, addr string, useTLS bool) *grpcLib.ClientConn { + t.Helper() + creds := insecure.NewCredentials() + if useTLS { + creds = credentials.NewTLS(&tls.Config{InsecureSkipVerify: true}) //nolint:gosec // ephemeral test certificate + } + conn, err := grpcLib.NewClient(addr, grpcLib.WithTransportCredentials(creds)) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +func casTestToken(t testing.TB, role casJWT.Role, maxBytes int64, expires time.Duration) string { + t.Helper() + builder, err := casJWT.NewBuilder(casJWT.WithPrivateKey("testdata/test-key.ec.pem"), casJWT.WithIssuer("cas-transport-test"), casJWT.WithExpiration(expires)) + require.NoError(t, err) + token, err := builder.GenerateJWT("test", "test-secret", casJWT.JWTAudience, role, maxBytes, "test-org") + require.NoError(t, err) + return token +} + +func casTestContext(token string) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + if token != "" { + ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+token) + } + return ctx, cancel +} + +func casResourceName(t testing.TB, resource *v1.CASResource) string { + t.Helper() + var buf bytes.Buffer + require.NoError(t, gob.NewEncoder(&buf).Encode(resource)) + return base64.StdEncoding.EncodeToString(buf.Bytes()) +} + +func casTestUpload(t testing.TB, client bytestream.ByteStreamClient, token, digest string, payload io.Reader, size int64) error { + t.Helper() + ctx, cancel := casTestContext(token) + defer cancel() + stream, err := client.Write(ctx) + if err != nil { + return err + } + resource := casResourceName(t, &v1.CASResource{Digest: digest, FileName: casTestFileName}) + chunk := make([]byte, 64<<10) + for { + n, readErr := payload.Read(chunk) + if readErr != nil && !errors.Is(readErr, io.EOF) { + return readErr + } + req := &bytestream.WriteRequest{ResourceName: resource, Data: chunk[:n], FinishWrite: errors.Is(readErr, io.EOF)} + if err = stream.Send(req); err != nil { + break + } + if errors.Is(readErr, io.EOF) { + break + } + } + resp, err := stream.CloseAndRecv() + if err == nil { + require.Equal(t, size, resp.GetCommittedSize()) + } + return err +} + +func TestSharedCASStatus(t *testing.T) { + sharedURL, legacy := startCASTestServers(t, casTestOpts(t)) + resp, err := http.Get(sharedURL + "/statusz") + require.NoError(t, err) + _, err = io.Copy(io.Discard, resp.Body) + require.NoError(t, resp.Body.Close()) + require.NoError(t, err) + require.Equal(t, http.StatusOK, resp.StatusCode) + for _, addr := range []string{sharedURL[len("http://"):], legacy} { + conn := casTestConn(t, addr, false) + ctx, cancel := casTestContext("") + _, err = v1.NewStatusServiceClient(conn).Statusz(ctx, &v1.StatuszRequest{}) + require.NoError(t, err) + stream, err := reflection.NewServerReflectionClient(conn).ServerReflectionInfo(ctx) + require.NoError(t, err) + require.NoError(t, stream.Send(&reflection.ServerReflectionRequest{MessageRequest: &reflection.ServerReflectionRequest_ListServices{}})) + result, err := stream.Recv() + require.NoError(t, err) + require.NotEmpty(t, result.GetListServicesResponse().GetService()) + require.NoError(t, stream.CloseSend()) + cancel() + } +} + +// TestCASProxyPreservesAuthority pins the proxy Rewrite contract: the native +// server must observe the client's original :authority, the same value a +// direct connection to the legacy listener would carry, so authority-based +// interceptors and telemetry do not see the loopback dial target. +func TestCASProxyPreservesAuthority(t *testing.T) { + opts := casTestOpts(t) + sharedURL, _ := startCASTestServers(t, opts) + host, port, err := net.SplitHostPort(sharedURL[len("http://"):]) + require.NoError(t, err) + + // Use an authority that differs from the dial address so a cleared Out.Host + // would be visible in the :authority the native server receives. + const authority = "cas.authority.test:443" + conn, err := grpcLib.NewClient(net.JoinHostPort(host, port), grpcLib.WithAuthority(authority), grpcLib.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + ctx, cancel := casTestContext("") + defer cancel() + _, err = v1.NewStatusServiceClient(conn).Statusz(ctx, &v1.StatuszRequest{}) + require.NoError(t, err) +} + +// TestCASProxyErrorTerminatesHTTP2Stream pins the ErrorHandler contract: on an +// upstream failure the proxy writes only the status header, and net/http +// finalizes the HTTP/2 response with END_STREAM when the handler returns, so +// the client gets a clean 502 instead of a stream left open. +func TestCASProxyErrorTerminatesHTTP2Stream(t *testing.T) { + proxy := grpcProxy("127.0.0.1:1", log.NewStdLogger(io.Discard)) // nothing listening + srv := httptest.NewUnstartedServer(proxy) + srv.EnableHTTP2 = true + srv.StartTLS() + defer srv.Close() + + tr := &http.Transport{Protocols: new(http.Protocols), TLSClientConfig: &tls.Config{InsecureSkipVerify: true}} //nolint:gosec // local test certificate + tr.Protocols.SetHTTP2(true) + defer tr.CloseIdleConnections() + client := &http.Client{Transport: tr, Timeout: 5 * time.Second} + + req, err := http.NewRequest(http.MethodPost, srv.URL+"/cas.v1.StatusService/Statusz", bytes.NewReader([]byte{0, 0, 0, 0, 0})) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/grpc") + req.Header.Set("Te", "trailers") + resp, err := client.Do(req) + require.NoError(t, err) + // Reading the body to completion must terminate, not wait for more frames. + _, err = io.Copy(io.Discard, resp.Body) + require.NoError(t, resp.Body.Close()) + require.NoError(t, err) + require.Equal(t, http.StatusBadGateway, resp.StatusCode) + require.Equal(t, 2, resp.ProtoMajor) +} + +// pacedUploads gates a ByteStream service so each direction can pause its +// consumer at a fixed transferred byte count. The hook stays inert until arm +// is closed (so a setup upload through the same service is not paused), and +// the gate wait is bounded: the hook runs on the server's stream goroutine, +// where an unbounded wait would deadlock cancellation, because the hook never +// sees the stream context. Production wiring never sets the hook. +func pacedUploads(t *testing.T, pauseAt int64) (arm func(), release chan struct{}, hook func(int64)) { + armed, release := make(chan struct{}), make(chan struct{}) + t.Cleanup(func() { close(release) }) + return func() { close(armed) }, release, func(transferred int64) { + select { + case <-armed: + default: + return + } + if transferred >= pauseAt { + select { + case <-release: + case <-time.After(3 * time.Second): + } + } + } +} + +// TestCASSharedPausedConsumerBackpressure is the durable regression for the +// read-ahead property: with the consumer paused mid-stream, the producer must +// stop advancing after bounded buffering, and cancellation must release both +// sides. A whole-stream buffer anywhere on the path (the measured failure of +// grpc.Server.ServeHTTP) would let the upload keep advancing to its full size. +func TestCASSharedPausedConsumerBackpressure(t *testing.T) { + const pauseAt = 8 << 20 // pause once 8 MiB have transferred + t.Run("upload", func(t *testing.T) { + opts := casTestOpts(t) + arm, release, hook := pacedUploads(t, pauseAt) + opts.ByteService = service.NewByteStreamService(opts.Providers, append(serviceOptsForTest(t), service.WithPaceHook(hook))...) + sharedURL, _ := startCASTestServers(t, opts) + client := bytestream.NewByteStreamClient(casTestConn(t, sharedURL[len("http://"):], false)) + up := casTestToken(t, casJWT.Uploader, 0, time.Hour) + + const size = 64 << 20 + hash := sha256.New() + _, err := io.CopyN(hash, zeroReader{}, size) + require.NoError(t, err) + digest := hex.EncodeToString(hash.Sum(nil)) + + arm() + ctx, cancel := casTestContext(up) + defer cancel() + stream, err := client.Write(ctx) + require.NoError(t, err) + resource := casResourceName(t, &v1.CASResource{Digest: digest, FileName: casTestFileName}) + chunk := make([]byte, 1<<20) + var sent atomic.Int64 + sendErr := make(chan error, 1) + go func() { + for sent.Load() < size { + n := min(int64(len(chunk)), size-sent.Load()) + err := stream.Send(&bytestream.WriteRequest{ResourceName: resource, Data: chunk[:n]}) + if err != nil { + sendErr <- err + return + } + sent.Add(n) + } + sendErr <- stream.CloseSend() + }() + + // Wait until the consumer has absorbed pauseAt bytes, then give the + // pipeline a moment: if any hop buffered whole streams, the client + // would keep advancing far past pauseAt. + require.Eventually(t, func() bool { return sent.Load() >= pauseAt }, 30*time.Second, 5*time.Millisecond) + time.Sleep(500 * time.Millisecond) + sentWhenPaused := sent.Load() + + // Cancellation must release the paused stream on both sides. + cancel() + select { + case err := <-sendErr: + require.Error(t, err, "the canceled upload must surface an error client-side") + case <-time.After(10 * time.Second): + require.FailNow(t, "cancellation did not release the paused upload") + } + + // Assert the stall only after the stream is torn down, so a failed + // assertion cannot leak the sender goroutine into cleanup. + require.Less(t, sentWhenPaused, int64(size)/4, + "an unbounded buffer let the producer advance past the paused consumer; sent %d bytes", sentWhenPaused) + _ = release // closed by t.Cleanup + }) + + t.Run("download", func(t *testing.T) { + opts := casTestOpts(t) + arm, release, hook := pacedUploads(t, pauseAt) + opts.ByteService = service.NewByteStreamService(opts.Providers, append(serviceOptsForTest(t), service.WithPaceHook(hook))...) + sharedURL, _ := startCASTestServers(t, opts) + client := bytestream.NewByteStreamClient(casTestConn(t, sharedURL[len("http://"):], false)) + up, down := casTestToken(t, casJWT.Uploader, 0, time.Hour), casTestToken(t, casJWT.Downloader, 0, time.Hour) + + const size = 64 << 20 + hash := sha256.New() + _, err := io.CopyN(hash, zeroReader{}, size) + require.NoError(t, err) + digest := hex.EncodeToString(hash.Sum(nil)) + // Upload the fixture with the gate still inert. + require.NoError(t, casTestUpload(t, client, up, digest, io.LimitReader(zeroReader{}, size), size)) + + arm() + ctx, cancel := casTestContext(down) + defer cancel() + stream, err := client.Read(ctx, &bytestream.ReadRequest{ResourceName: digest}) + require.NoError(t, err) + var received atomic.Int64 + recvErr := make(chan error, 1) + go func() { + for { + resp, err := stream.Recv() + if err != nil { + recvErr <- err + return + } + received.Add(int64(len(resp.GetData()))) + } + }() + + require.Eventually(t, func() bool { return received.Load() >= pauseAt }, 30*time.Second, 5*time.Millisecond) + time.Sleep(500 * time.Millisecond) + receivedWhenPaused := received.Load() + + cancel() + select { + case <-recvErr: + case <-time.After(10 * time.Second): + require.FailNow(t, "cancellation did not release the paused download") + } + + require.Less(t, receivedWhenPaused, int64(size)/4, + "an unbounded buffer let the server stream past the paused consumer; received %d bytes", receivedWhenPaused) + _ = release // closed by t.Cleanup + }) +} + +// serviceOptsForTest mirrors the test wiring in casTestOpts so pace-hook +// variants get the same staging option. +func serviceOptsForTest(tb testing.TB) []service.NewOpt { + return []service.NewOpt{service.WithStagingDir(tb.TempDir())} +} + +func TestCASByteStreamTransportParity(t *testing.T) { + for _, useTLS := range []bool{false, true} { + t.Run(map[bool]string{false: "h2c", true: "TLS"}[useTLS], func(t *testing.T) { + for _, endpoint := range []string{casSharedEndpoint, casLegacyEndpoint} { + t.Run(endpoint, func(t *testing.T) { + opts := casTestOpts(t) + if useTLS { + opts.ServerConfig.Http.TlsConfig = casTestTLS(t) + opts.ServerConfig.Grpc.TlsConfig = casTestTLS(t) + } + sharedURL, legacy := startCASTestServers(t, opts) + addr := legacy + if endpoint == casSharedEndpoint { + addr = sharedURL[len("http://"):] + if useTLS { + addr = sharedURL[len("https://"):] + } + } + conn := casTestConn(t, addr, useTLS) + client := bytestream.NewByteStreamClient(conn) + content := bytes.Repeat([]byte("streamed artifact\n"), 64<<10) + sum := sha256.Sum256(content) + digest := hex.EncodeToString(sum[:]) + up := casTestToken(t, casJWT.Uploader, 0, time.Hour) + down := casTestToken(t, casJWT.Downloader, 0, time.Hour) + require.NoError(t, casTestUpload(t, client, up, digest, bytes.NewReader(content), int64(len(content)))) + ctx, cancel := casTestContext(down) + defer cancel() + stream, err := client.Read(ctx, &bytestream.ReadRequest{ResourceName: digest}) + require.NoError(t, err) + hash := sha256.New() + var size int64 + for { + msg, err := stream.Recv() + if errors.Is(err, io.EOF) { + break + } + require.NoError(t, err) + n, err := hash.Write(msg.GetData()) + require.NoError(t, err) + size += int64(n) + } + require.EqualValues(t, len(content), size) + require.Equal(t, digest, hex.EncodeToString(hash.Sum(nil))) + _, err = v1.NewResourceServiceClient(conn).Describe(ctx, &v1.ResourceServiceDescribeRequest{Digest: digest}) + require.NoError(t, err) + _, err = v1.NewResourceServiceClient(conn).Describe(ctx, &v1.ResourceServiceDescribeRequest{}) + require.Equal(t, codes.InvalidArgument, status.Code(err)) + + // The HTTP download remains usable regardless of which RPC listener uploaded the blob. + // The download uses HTTP/1.1: the shared endpoint serves + // plain HTTP over HTTP/1.1 and HTTP/2 (D-020), and the + // legacy behavior over HTTP/1.1 must keep working. + tr := &http.Transport{Protocols: new(http.Protocols), TLSClientConfig: &tls.Config{InsecureSkipVerify: true}} //nolint:gosec // test certificate + tr.Protocols.SetHTTP1(true) + defer tr.CloseIdleConnections() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, sharedURL+"/download/sha256:"+digest, nil) + require.NoError(t, err) + req.Header.Set("Authorization", "Bearer "+down) + response, err := (&http.Client{Transport: tr, Timeout: 5 * time.Second}).Do(req) + require.NoError(t, err) + require.Equal(t, http.StatusOK, response.StatusCode) + hash.Reset() + size, err = io.Copy(hash, response.Body) + require.NoError(t, response.Body.Close()) + require.NoError(t, err) + require.EqualValues(t, len(content), size) + require.Equal(t, digest, hex.EncodeToString(hash.Sum(nil))) + }) + } + }) + } +} + +func casTestTLS(t testing.TB) *conf.Server_TLS { + t.Helper() + srv := httptest.NewTLSServer(http.NotFoundHandler()) + defer srv.Close() + pair := srv.TLS.Certificates[0] + key, err := x509.MarshalPKCS8PrivateKey(pair.PrivateKey) + require.NoError(t, err) + dir := t.TempDir() + certPath, keyPath := filepath.Join(dir, "cert.pem"), filepath.Join(dir, "key.pem") + require.NoError(t, os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: pair.Certificate[0]}), 0600)) + require.NoError(t, os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: key}), 0600)) + return &conf.Server_TLS{Certificate: certPath, PrivateKey: keyPath} +} + +// Native gRPC and ordinary HTTP must work through the shared port (D-020). +func TestCASSharedProtocols(t *testing.T) { + for _, useTLS := range []bool{false, true} { + t.Run(map[bool]string{false: "h2c", true: "TLS"}[useTLS], func(t *testing.T) { + opts := casTestOpts(t) + if useTLS { + opts.ServerConfig.Http.TlsConfig = casTestTLS(t) + } + sharedURL, _ := startCASTestServers(t, opts) + + var connections atomic.Int32 + // One catch-all upstream connection that carries both protocols, as + // a single-location ingress with an h2c (or TLS h2) upstream sends + // them. HTTP/1 must stay off in the client: net/http only uses h2c + // prior knowledge when HTTP/1 is disabled. + tr := &http.Transport{Protocols: new(http.Protocols), TLSClientConfig: &tls.Config{InsecureSkipVerify: true}} //nolint:gosec // test certificate + tr.Protocols.SetHTTP2(true) + tr.Protocols.SetUnencryptedHTTP2(true) + tr.Protocols.SetHTTP1(false) + tr.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + connections.Add(1) + return (&net.Dialer{}).DialContext(ctx, network, addr) + } + defer tr.CloseIdleConnections() + client := &http.Client{Transport: tr, Timeout: 5 * time.Second} + + perform := func(kind string) { + path, method := "/cas.v1.StatusService/Statusz", http.MethodPost + if kind == casHTTPRequest { + path, method = "/statusz", http.MethodGet + } + if kind == casPreflightRequest { + path, method = "/download/sha256:abc", http.MethodOptions + } + req, err := http.NewRequest(method, sharedURL+path, bytes.NewReader([]byte{0, 0, 0, 0, 0})) + require.NoError(t, err) + if kind == casGRPCRequest { + req.Header.Set("Content-Type", "application/grpc") + req.Header.Set("Te", "trailers") + } + if kind == casPreflightRequest { + req.Header.Set("Origin", "https://example.test") + req.Header.Set("Access-Control-Request-Method", "GET") + } + resp, err := client.Do(req) + require.NoError(t, err) + _, err = io.Copy(io.Discard, resp.Body) + require.NoError(t, resp.Body.Close()) + require.NoError(t, err) + expected := http.StatusOK + if kind == casPreflightRequest { + expected = http.StatusNoContent + } + require.Equal(t, expected, resp.StatusCode, kind) + if kind == casGRPCRequest { + // Native gRPC requires HTTP/2 and its trailers round-trip + // through the reverse proxy to the private native server. + require.Equal(t, 2, resp.ProtoMajor) + require.Equal(t, "0", resp.Trailer.Get("Grpc-Status")) + } + } + + // D-020: the shared endpoint dispatches each request, so native + // gRPC and ordinary HTTP interleave in any order — including on + // one pooled upstream HTTP/2 connection, as a single-location + // ingress sends them. + perform(casGRPCRequest) + perform(casHTTPRequest) + perform(casGRPCRequest) + perform(casPreflightRequest) + perform(casHTTPRequest) + perform(casGRPCRequest) + + // All six requests reused one HTTP/2 connection. + require.EqualValues(t, 1, connections.Load()) + }) + } +} + +// These focused cases compare the existing middleware on both transports; +// they do not replace the complete original R-002 qualification matrix. +func TestCASAuthenticationAndUploadErrors(t *testing.T) { + for _, endpoint := range []string{casSharedEndpoint, casLegacyEndpoint} { + t.Run(endpoint, func(t *testing.T) { + opts := casTestOpts(t) + sharedURL, legacy := startCASTestServers(t, opts) + addr := legacy + if endpoint == casSharedEndpoint { + addr = sharedURL[len("http://"):] + } + conn := casTestConn(t, addr, false) + client := bytestream.NewByteStreamClient(conn) + content := []byte("an authenticated CAS artifact") + sum := sha256.Sum256(content) + digest := hex.EncodeToString(sum[:]) + tests := []struct { + name, token, digest string + code codes.Code + }{ + {name: "missing token", digest: digest, code: codes.Unauthenticated}, + {name: "malformed token", token: "invalid.jwt.token", digest: digest, code: codes.Unauthenticated}, + {name: casExpiredTokenCase, token: casTestToken(t, casJWT.Uploader, 0, -time.Hour), digest: digest, code: codes.Unauthenticated}, + {name: "downloader cannot upload", token: casTestToken(t, casJWT.Downloader, 0, time.Hour), digest: digest, code: codes.Unauthenticated}, + {name: "upload size cap", token: casTestToken(t, casJWT.Uploader, 1, time.Hour), digest: digest, code: codes.ResourceExhausted}, + {name: "digest mismatch", token: casTestToken(t, casJWT.Uploader, 0, time.Hour), digest: hex.EncodeToString(make([]byte, sha256.Size)), code: codes.InvalidArgument}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := casTestUpload(t, client, tc.token, tc.digest, bytes.NewReader(content), int64(len(content))) + require.Equal(t, tc.code, status.Code(err), "%v", err) + }) + } + for _, tc := range []struct { + name, token string + code codes.Code + }{ + {name: "missing token", code: codes.Unauthenticated}, + {name: casExpiredTokenCase, token: casTestToken(t, casJWT.Downloader, 0, -time.Hour), code: codes.Unauthenticated}, + {name: "uploader cannot download", token: casTestToken(t, casJWT.Uploader, 0, time.Hour), code: codes.Unauthenticated}, + } { + t.Run("read/"+tc.name, func(t *testing.T) { + ctx, cancel := casTestContext(tc.token) + defer cancel() + stream, err := client.Read(ctx, &bytestream.ReadRequest{ResourceName: digest}) + require.NoError(t, err) + _, err = stream.Recv() + require.Equal(t, tc.code, status.Code(err), "%v", err) + }) + } + }) + } +} + +// TestCASLegacyListenerDeprecationWarning covers the R-006 startup warning: +// the shared endpoint logs a deprecation warning when the legacy gRPC +// listener is enabled, matching the control plane. +func TestCASLegacyListenerDeprecationWarning(t *testing.T) { + for _, tc := range []struct { + name string + enabled *bool + wantLog bool + }{ + {name: "default on", wantLog: true}, + {name: "explicit on", enabled: new(true), wantLog: true}, + {name: "explicit off", enabled: new(false), wantLog: false}, + } { + t.Run(tc.name, func(t *testing.T) { + opts := casTestOpts(t) + opts.ServerConfig.LegacyGrpcListener = tc.enabled + var buf bytes.Buffer + opts.Logger = log.NewStdLogger(&buf) + servers, err := NewGRPCServers(opts) + require.NoError(t, err) + api, err := NewHTTPServer(opts, servers) + require.NoError(t, err) + require.Equal(t, tc.wantLog, strings.Contains(buf.String(), "the gRPC-only listener"), "logs: %s", buf.String()) + require.NoError(t, api.Stop(context.Background())) + }) + } +} + +func TestCASLegacyListenerAndTLSConfiguration(t *testing.T) { + for _, tc := range []struct { + name string + enabled *bool + wantLegacy bool + }{ + {name: "default", wantLegacy: true}, + {name: "explicit on", enabled: new(true), wantLegacy: true}, + {name: "explicit off", enabled: new(false)}, + } { + t.Run(tc.name, func(t *testing.T) { + opts := casTestOpts(t) + opts.ServerConfig.LegacyGrpcListener = tc.enabled + if !tc.wantLegacy { + opts.ServerConfig.Grpc.TlsConfig = &conf.Server_TLS{Certificate: "unused-invalid-cert", PrivateKey: "unused-invalid-key"} + } + sharedURL, legacy := startCASTestServers(t, opts) + require.Equal(t, tc.wantLegacy, legacy != "") + conn := casTestConn(t, sharedURL[len("http://"):], false) + ctx, cancel := casTestContext("") + defer cancel() + _, err := v1.NewStatusServiceClient(conn).Statusz(ctx, &v1.StatuszRequest{}) + require.NoError(t, err) + }) + } + for _, tc := range []struct { + name string + http *conf.Server_HTTP + wantErr string + }{ + {name: "missing shared address", http: &conf.Server_HTTP{}, wantErr: "server.http.addr must be set"}, + {name: "partial shared TLS", http: &conf.Server_HTTP{Addr: casAnyAddr, TlsConfig: &conf.Server_TLS{Certificate: "only-cert"}}, wantErr: "incomplete TLS configuration"}, + } { + t.Run(tc.name, func(t *testing.T) { + opts := casTestOpts(t) + opts.ServerConfig.Http = tc.http + opts.ServerConfig.LegacyGrpcListener = new(false) + servers, err := NewGRPCServers(opts) + require.NoError(t, err) + // A failed NewHTTPServer must close the private socket itself. + _, err = NewHTTPServer(opts, servers) + require.ErrorContains(t, err, tc.wantErr) + }) + } +} + +func TestCASCancelledUpload(t *testing.T) { + for _, endpoint := range []string{casSharedEndpoint, casLegacyEndpoint} { + t.Run(endpoint, func(t *testing.T) { + opts := casTestOpts(t) + sharedURL, legacy := startCASTestServers(t, opts) + addr := legacy + if endpoint == casSharedEndpoint { + addr = sharedURL[len("http://"):] + } + client := bytestream.NewByteStreamClient(casTestConn(t, addr, false)) + ctx, cancel := casTestContext(casTestToken(t, casJWT.Uploader, 0, time.Hour)) + defer cancel() + stream, err := client.Write(ctx) + require.NoError(t, err) + digest := hex.EncodeToString(make([]byte, sha256.Size)) + require.NoError(t, stream.Send(&bytestream.WriteRequest{ResourceName: casResourceName(t, &v1.CASResource{Digest: digest}), Data: make([]byte, 64<<10)})) + cancel() + _, err = stream.CloseAndRecv() + require.Equal(t, codes.Canceled, status.Code(err)) + // No partial blob reaches the backend after cancellation. + p, err := opts.Providers["test"].FromCredentials(context.Background(), "test-secret") + require.NoError(t, err) + require.Eventually(t, func() bool { + exists, err := p.Exists(context.Background(), digest) + return err == nil && !exists + }, time.Second, 10*time.Millisecond) + }) + } +} + +// TestCASSharedConcurrentUploadsMemoryBounded pushes several large streams +// through the shared endpoint at once, multiplexed on one connection as a +// single-location ingress sends them. The payload is generated on the fly, so +// no test-owned buffer is resident. Each hop must window-pace the data: if +// any hop buffered whole streams instead, the sampled peak heap would reach +// the total transferred bytes. The single-stream benchmark showed the pacing +// directly (about 32 MiB peak for a 1 GiB upload, against about 1.25 GiB on +// the rejected grpc.Server.ServeHTTP adapter under the same conditions); this +// test adds the concurrent-stream dimension. +func TestCASSharedConcurrentUploadsMemoryBounded(t *testing.T) { + const ( + streams = 4 + size = 128 << 20 // 128 MiB per stream, 512 MiB total + ) + opts := casTestOpts(t) + sharedURL, _ := startCASTestServers(t, opts) + client := bytestream.NewByteStreamClient(casTestConn(t, sharedURL[len("http://"):], false)) + up := casTestToken(t, casJWT.Uploader, 0, time.Hour) + + hash := sha256.New() + _, err := io.CopyN(hash, zeroReader{}, size) + require.NoError(t, err) + digest := hex.EncodeToString(hash.Sum(nil)) + + stopSample := sampleCASHeap() + var wg sync.WaitGroup + errs := make(chan error, streams) + for range streams { + wg.Add(1) + go func() { + defer wg.Done() + errs <- casTestUpload(t, client, up, digest, io.LimitReader(zeroReader{}, size), size) + }() + } + wg.Wait() + close(errs) + for err := range errs { + require.NoError(t, err) + } + peak := stopSample() + t.Logf("peak heap: %d MiB for %d MiB across %d streams", peak>>20, (size*streams)>>20, streams) + + // Whole-stream buffering on any hop would push the peak toward the + // transferred bytes; window pacing keeps it at per-stream buffers. + require.Less(t, peak, uint64(300<<20), + "four concurrent 128 MiB uploads must not buffer whole streams; peak heap %d bytes", peak) +} + +// TestCASSharedDualTransportFailureDoesNotLeak proves the errCh contract: when +// both transports fail before Start reads the first error, the second +// transport's report is still buffered (capacity equals the worker count), so +// its goroutine finishes and is not leaked. Stop's workers.Wait() would hang +// if the second send blocked. +func TestCASSharedDualTransportFailureDoesNotLeak(t *testing.T) { + opts := casTestOpts(t) + opts.ServerConfig.LegacyGrpcListener = new(false) + servers, err := NewGRPCServers(opts) + require.NoError(t, err) + api, err := NewHTTPServer(opts, servers) + require.NoError(t, err) + // Fail both transports before Start: close the shared HTTP listener and + // the private gRPC listener. + require.NoError(t, api.lis.Close()) + require.NoError(t, servers.Close()) + + done := make(chan error, 1) + go func() { done <- api.Start(context.Background()) }() + select { + case err := <-done: + require.Error(t, err, "Start must fail when both transports fail") + case <-time.After(10 * time.Second): + require.FailNow(t, "Start did not return on dual-transport failure") + } + // The second writer must not be stuck: Stop would otherwise block forever + // in workers.Wait(). + stopped := make(chan struct{}) + go func() { _ = api.Stop(context.Background()); close(stopped) }() + select { + case <-stopped: + case <-time.After(10 * time.Second): + require.FailNow(t, "Stop hung: a transport goroutine is still blocked on errCh") + } +} + +func TestCASSharedShutdownDrainsUpload(t *testing.T) { + opts := casTestOpts(t) + opts.ServerConfig.LegacyGrpcListener = new(false) + provider := opts.Providers["test"].(*mocks.Provider) + entered, release := make(chan struct{}), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + provider.ExpectedCalls[0].Run(func(mock.Arguments) { close(entered); <-release }) + servers, err := NewGRPCServers(opts) + require.NoError(t, err) + api, err := NewHTTPServer(opts, servers) + require.NoError(t, err) + endpoint, err := api.Endpoint() + require.NoError(t, err) + startCASTestTransport(t, api) + client := bytestream.NewByteStreamClient(casTestConn(t, endpoint.Host, false)) + ctx, cancel := casTestContext(casTestToken(t, casJWT.Uploader, 0, time.Hour)) + defer cancel() + content := make([]byte, 64<<10) + sum := sha256.Sum256(content) + stream, err := client.Write(ctx) + require.NoError(t, err) + require.NoError(t, stream.Send(&bytestream.WriteRequest{ResourceName: casResourceName(t, &v1.CASResource{Digest: hex.EncodeToString(sum[:])}), Data: content})) + select { + case <-entered: + case <-ctx.Done(): + require.FailNow(t, "upload did not reach the backend lookup") + } + stopped := make(chan error, 1) + go func() { + stopCtx, stopCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer stopCancel() + stopped <- api.Stop(stopCtx) + }() + select { + case err := <-stopped: + require.FailNow(t, "shutdown returned before the upload completed", "%v", err) + case <-time.After(50 * time.Millisecond): + } + unblock() + require.NoError(t, stream.Send(&bytestream.WriteRequest{ResourceName: casResourceName(t, &v1.CASResource{Digest: hex.EncodeToString(sum[:])}), FinishWrite: true})) + response, err := stream.CloseAndRecv() + require.NoError(t, err) + require.EqualValues(t, len(content), response.GetCommittedSize()) + select { + case err := <-stopped: + require.NoError(t, err) + case <-ctx.Done(): + require.FailNow(t, "shutdown did not complete after the upload") + } +} + +// TestCASSharedGRPCFailureStopsEndpoint: if the private gRPC transport fails, +// the shared endpoint must not keep serving HTTP while proxied gRPC answers +// 502. Start must surface the failure so the app stops. +func TestCASSharedGRPCFailureStopsEndpoint(t *testing.T) { + opts := casTestOpts(t) + opts.ServerConfig.LegacyGrpcListener = new(false) + servers, err := NewGRPCServers(opts) + require.NoError(t, err) + api, err := NewHTTPServer(opts, servers) + require.NoError(t, err) + // Simulate the private gRPC transport failing: close its listener before + // the servers start. grpc-go's Serve on the closed listener returns an + // error that Start must not swallow. + require.NoError(t, servers.Close()) + startErr := make(chan error, 1) + go func() { startErr <- api.Start(context.Background()) }() + select { + case err := <-startErr: + require.Error(t, err, "a failed private gRPC transport must fail the shared endpoint") + case <-time.After(10 * time.Second): + require.FailNow(t, "Start kept serving after the private gRPC transport failed") + } + // Best-effort teardown of the HTTP server that started on the shared port. + _ = api.Stop(context.Background()) +} + +func TestCASSharedInitializationDoesNotBindOnInvalidKey(t *testing.T) { + opts := casTestOpts(t) + opts.ServerConfig.LegacyGrpcListener = new(false) + servers, err := NewGRPCServers(opts) + require.NoError(t, err) + defer func() { require.NoError(t, servers.Private.Stop(context.Background())) }() + reserved, err := net.Listen("tcp", casAnyAddr) + require.NoError(t, err) + opts.ServerConfig.Http.Addr = reserved.Addr().String() + require.NoError(t, reserved.Close()) + opts.AuthConfig.PublicKeyPath = "testdata/missing-public-key" + _, err = NewHTTPServer(opts, servers) + require.Error(t, err) + retry, err := net.Listen("tcp", opts.ServerConfig.Http.Addr) + require.NoError(t, err, "failed initialization must leave the listener address available") + require.NoError(t, retry.Close()) + // The failed constructor closed the private socket: connections are refused. + require.NoError(t, servers.Close(), "Close after a failed construction is safe") + _, err = net.DialTimeout("tcp", servers.PrivateAddr, time.Second) + require.Error(t, err, "the private socket must refuse connections after the failed construction") +} + +// TestCASPrivateSocketClosedWhenNeverStarted covers the stop-before-serve +// path: the socket must be closable and stop accepting after Close, even +// though Private.Stop alone does not close a listener registered through +// grpc.Listener before Serve runs. +func TestCASPrivateSocketClosedWhenNeverStarted(t *testing.T) { + opts := casTestOpts(t) + opts.ServerConfig.LegacyGrpcListener = new(false) + servers, err := NewGRPCServers(opts) + require.NoError(t, err) + require.NoError(t, servers.Close()) + _, err = net.DialTimeout("tcp", servers.PrivateAddr, time.Second) + require.Error(t, err, "the private socket must refuse connections after Close") + require.NoError(t, servers.Close(), "Close is idempotent") +} diff --git a/app/artifact-cas/internal/server/httpmetrics.go b/app/artifact-cas/internal/server/httpmetrics.go index b37b58576..841554684 100644 --- a/app/artifact-cas/internal/server/httpmetrics.go +++ b/app/artifact-cas/internal/server/httpmetrics.go @@ -1,5 +1,5 @@ // -// Copyright 2023 The Chainloop Authors. +// Copyright 2023-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package server import ( "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" + "github.com/go-kratos/kratos/v2/log" "github.com/go-kratos/kratos/v2/middleware/recovery" "github.com/go-kratos/kratos/v2/transport/http" "github.com/prometheus/client_golang/prometheus/promhttp" @@ -27,8 +28,13 @@ type HTTPMetricsServer struct { *http.Server } -// NewHTTPMetricsServer exposes the metrics endpoint -func NewHTTPMetricsServer(c *conf.Server) (*HTTPMetricsServer, error) { +// NewHTTPMetricsServer exposes the metrics endpoint. server.http_metrics never +// applied TLS: a configured tls_config is ignored with a warning rather than +// rejected, so existing deployments that set it keep working. +func NewHTTPMetricsServer(c *conf.Server, logger log.Logger) (*HTTPMetricsServer, error) { + if c.GetHttpMetrics().GetTlsConfig() != nil { + log.NewHelper(logger).Warn("server.http_metrics.tls_config is ignored: metrics stay plaintext; configure shared endpoint TLS under server.http.tls_config") + } var opts = []http.ServerOption{ http.Middleware( recovery.Recovery(), diff --git a/app/artifact-cas/internal/server/httpmetrics_test.go b/app/artifact-cas/internal/server/httpmetrics_test.go new file mode 100644 index 000000000..9c6efa4d4 --- /dev/null +++ b/app/artifact-cas/internal/server/httpmetrics_test.go @@ -0,0 +1,43 @@ +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "io" + "testing" + + "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" + "github.com/go-kratos/kratos/v2/log" + "github.com/stretchr/testify/require" +) + +func TestCASMetricsTLSConfiguration(t *testing.T) { + logger := log.NewStdLogger(io.Discard) + for _, tc := range []struct { + name string + tls *conf.Server_TLS + }{ + {name: "plaintext"}, + {name: "empty TLS config", tls: &conf.Server_TLS{}}, + {name: "TLS config", tls: &conf.Server_TLS{Certificate: "cert.pem", PrivateKey: "key.pem"}}, + } { + t.Run(tc.name, func(t *testing.T) { + // A configured but unsupported tls_config is ignored with a warning, + // so existing deployments that set it keep working. + _, err := NewHTTPMetricsServer(&conf.Server{HttpMetrics: &conf.Server_HTTP{Addr: casAnyAddr, TlsConfig: tc.tls}}, logger) + require.NoError(t, err) + }) + } +} diff --git a/app/artifact-cas/internal/server/server.go b/app/artifact-cas/internal/server/server.go index 73b2b2280..4897ffc24 100644 --- a/app/artifact-cas/internal/server/server.go +++ b/app/artifact-cas/internal/server/server.go @@ -1,5 +1,5 @@ // -// Copyright 2023 The Chainloop Authors. +// Copyright 2023-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -20,6 +20,6 @@ import ( ) // ProviderSet is server providers. -var ProviderSet = wire.NewSet(NewGRPCServer, NewHTTPServer, NewHTTPMetricsServer, NewTracerProvider) +var ProviderSet = wire.NewSet(wire.Struct(new(Opts), "*"), NewGRPCServers, NewHTTPServer, NewHTTPMetricsServer, NewTracerProvider) var Version = "dev" diff --git a/app/artifact-cas/internal/server/streaming_bench_test.go b/app/artifact-cas/internal/server/streaming_bench_test.go new file mode 100644 index 000000000..10b884671 --- /dev/null +++ b/app/artifact-cas/internal/server/streaming_bench_test.go @@ -0,0 +1,151 @@ +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "os" + "path/filepath" + "runtime" + "strconv" + "sync" + "testing" + "time" + + casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" + "github.com/stretchr/testify/require" + "google.golang.org/genproto/googleapis/bytestream" +) + +// This compares the production CAS service and interceptors over the unchanged +// legacy listener and the shared endpoint (per-request dispatch with the reverse +// proxy to the private native server), both on the native transport +// with the same disk-backed test backend. +// It is a transport experiment, not the original R-003 release/S3 qualification. +// Use a disk-backed TMPDIR: /tmp may be tmpfs on development machines. +func BenchmarkCASByteStreamTransport(b *testing.B) { + size := int64(1 << 30) + if value := os.Getenv("CAS_BENCH_BYTES"); value != "" { + parsed, err := strconv.ParseInt(value, 10, 64) + require.NoError(b, err) + require.Positive(b, parsed) + size = parsed + } + hash := sha256.New() + _, err := io.CopyN(hash, zeroReader{}, size) + require.NoError(b, err) + digest := hex.EncodeToString(hash.Sum(nil)) + for _, endpoint := range []string{casLegacyEndpoint, casSharedEndpoint} { + b.Run(endpoint, func(b *testing.B) { + opts := casTestOpts(b) + sharedURL, legacy := startCASTestServers(b, opts) + addr := legacy + if endpoint == casSharedEndpoint { + addr = sharedURL[len("http://"):] + } + client := bytestream.NewByteStreamClient(casTestConn(b, addr, false)) + storage, err := opts.Providers["test"].FromCredentials(context.Background(), "test-secret") + require.NoError(b, err) + path := filepath.Join(storage.(*diskBackend).dir, digest) + up := casTestToken(b, casJWT.Uploader, 0, time.Hour) + down := casTestToken(b, casJWT.Downloader, 0, time.Hour) + b.Run("upload", func(b *testing.B) { + b.SetBytes(size) + b.ReportAllocs() + stopSample := sampleCASHeap() + defer stopSample() + b.ResetTimer() + for range b.N { + b.StopTimer() + err := os.Remove(path) + require.True(b, err == nil || os.IsNotExist(err), "%v", err) + b.StartTimer() + require.NoError(b, casTestUpload(b, client, up, digest, io.LimitReader(zeroReader{}, size), size)) + } + b.StopTimer() + b.ReportMetric(float64(stopSample())/(1<<20), "peak-heap-MiB") + }) + // Seed downloads independently of the upload benchmark. A sparse + // file represents the same zero bytes without a heap-sized buffer. + f, err := os.Create(path) + require.NoError(b, err) + require.NoError(b, f.Truncate(size)) + require.NoError(b, f.Close()) + b.Run("download", func(b *testing.B) { + b.SetBytes(size) + b.ReportAllocs() + stopSample := sampleCASHeap() + defer stopSample() + b.ResetTimer() + for range b.N { + ctx, cancel := casTestContext(down) + stream, err := client.Read(ctx, &bytestream.ReadRequest{ResourceName: digest}) + require.NoError(b, err) + hash := sha256.New() + var received int64 + for { + msg, err := stream.Recv() + if errors.Is(err, io.EOF) { + break + } + require.NoError(b, err) + n, err := hash.Write(msg.GetData()) + require.NoError(b, err) + received += int64(n) + } + cancel() + require.Equal(b, size, received) + require.Equal(b, digest, hex.EncodeToString(hash.Sum(nil))) + } + b.StopTimer() + b.ReportMetric(float64(stopSample())/(1<<20), "peak-heap-MiB") + }) + }) + } +} + +type zeroReader struct{} + +func (zeroReader) Read(p []byte) (int, error) { clear(p); return len(p), nil } + +// sampleCASHeap reports sampled live heap, not total allocated bytes. Both the +// client and server run here; this is not a server-only RSS or bounded-memory proof. +func sampleCASHeap() func() uint64 { + runtime.GC() + done, result := make(chan struct{}), make(chan uint64, 1) + go func() { + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + var peak uint64 + for { + var stats runtime.MemStats + runtime.ReadMemStats(&stats) + if stats.HeapAlloc > peak { + peak = stats.HeapAlloc + } + select { + case <-done: + result <- peak + return + case <-ticker.C: + } + } + }() + return sync.OnceValue(func() uint64 { close(done); return <-result }) +} diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go index 01d0d42c4..38d6ee77a 100644 --- a/app/artifact-cas/internal/service/bytestream.go +++ b/app/artifact-cas/internal/service/bytestream.go @@ -216,7 +216,7 @@ func (s *ByteStreamService) dispatchSkippedUpload(info *casJWT.Claims, resource // returned unwrapped for the caller to classify. func (s *ByteStreamService) spillVerifyUpload(ctx context.Context, stream bytestream.ByteStream_WriteServer, storageBackend backend.Uploader, req *writeRequest, maxBytes int64) (int64, error) { f, size, err := s.stageAndVerify(stagingUpload, req.resource.Digest, func(w io.Writer) error { - return spillStream(ctx, stream, w, req.GetData(), maxBytes, s.log, req.resource.Digest) + return spillStream(ctx, stream, w, req.GetData(), maxBytes, s.log, req.resource.Digest, s.paceHook) }) if err != nil { return 0, err @@ -246,8 +246,10 @@ func (e *backendUploadError) Unwrap() error { return e.err } // spillStream forwards the artifact from the client stream into w (the staging // file tee'd into a SHA256 hasher), enforcing the max upload size as it goes. -// firstData is the payload already read from the first request. -func spillStream(ctx context.Context, stream bytestream.ByteStream_WriteServer, w io.Writer, firstData []byte, maxSize int64, log *log.Helper, digest string) error { +// firstData is the payload already read from the first request. paceHook, when +// non-nil, observes each chunk's cumulative size; production wiring leaves it +// nil (see commonService.paceHook). +func spillStream(ctx context.Context, stream bytestream.ByteStream_WriteServer, w io.Writer, firstData []byte, maxSize int64, log *log.Helper, digest string, paceHook func(transferred int64)) error { // running total, needed to enforce the cap before each write var size int64 write := func(data []byte) error { @@ -255,6 +257,9 @@ func spillStream(ctx context.Context, stream bytestream.ByteStream_WriteServer, return nil } size += int64(len(data)) + if paceHook != nil { + paceHook(size) + } if err := checkUploadSize(size, maxSize); err != nil { return err } @@ -344,7 +349,7 @@ func (s *ByteStreamService) Read(req *bytestream.ReadRequest, stream bytestream. } defer s.closeStagingFile(f) - if err := copyStaged(sendWriter{stream}, f); err != nil { + if err := copyStaged(&pacedWriter{w: sendWriter{stream}, hook: s.paceHook}, f); err != nil { return s.downloadError(ctx, err, req.ResourceName) } @@ -434,3 +439,20 @@ type sendWriter struct { func (sw sendWriter) Write(data []byte) (int, error) { return len(data), sw.stream.Send(&bytestream.ReadResponse{Data: data}) } + +// pacedWriter reports each write's cumulative bytes to a pacing hook. When the +// hook is nil it forwards untouched, which is the production case. +type pacedWriter struct { + w io.Writer + hook func(transferred int64) + cumSum int64 +} + +func (p *pacedWriter) Write(data []byte) (int, error) { + n, err := p.w.Write(data) + p.cumSum += int64(n) + if err == nil && p.hook != nil { + p.hook(p.cumSum) + } + return n, err +} diff --git a/app/artifact-cas/internal/service/service.go b/app/artifact-cas/internal/service/service.go index 87e69f0e8..f60c8d900 100644 --- a/app/artifact-cas/internal/service/service.go +++ b/app/artifact-cas/internal/service/service.go @@ -51,6 +51,11 @@ type commonService struct { existence cache.Cache[casexistence.Entry] // clients caches the backend clients loaded for uploads, nil when disabled clients *expirable.LRU[string, backend.UploaderDownloader] + // paceHook, when set, is called once per transferred chunk (uploads and + // downloads) with the bytes transferred so far. Production wiring never + // sets it; transfer-pace tests use it to stall the consumer mid-stream and + // observe the end-to-end HTTP/2 backpressure through the shared endpoint. + paceHook func(transferred int64) } func (s *commonService) loadBackend(ctx context.Context, providerType, secretID string) (backend.UploaderDownloader, error) { @@ -95,6 +100,15 @@ func WithStagingDir(dir string) NewOpt { } } +// WithPaceHook installs a per-chunk pacing callback for transfer-pace tests. +// Production wiring never sets it: the callback runs on every chunk of every +// transfer, so anything but a nil default would sit on the streaming hot path. +func WithPaceHook(hook func(transferred int64)) NewOpt { + return func(s *commonService) { + s.paceHook = hook + } +} + func newCommonService(backends backend.Providers, opts ...NewOpt) *commonService { s := &commonService{ log: servicelogger.EmptyLogger(), diff --git a/deployment/chainloop/Chart.yaml b/deployment/chainloop/Chart.yaml index 713c92fd9..f10ac5164 100644 --- a/deployment/chainloop/Chart.yaml +++ b/deployment/chainloop/Chart.yaml @@ -7,7 +7,7 @@ description: Chainloop is an open source software supply chain control plane, a type: application # Bump the patch (not minor, not major) version on each change in the Chart Source code -version: 1.468.1 +version: 1.468.3 # Do not update appVersion, this is handled automatically by the release process appVersion: v1.125.0 diff --git a/deployment/chainloop/README.md b/deployment/chainloop/README.md index 28d51d8d6..f1e2e03a2 100644 --- a/deployment/chainloop/README.md +++ b/deployment/chainloop/README.md @@ -245,7 +245,25 @@ The control plane ingress serves HTTP, gRPC-Web, and native gRPC on the same hos Its HTTP server routes each request, so these protocols can share an upstream HTTP/2 connection. The existing gRPC ingress stays available. Existing clients can keep their current endpoints. To test the shared endpoint, point a gRPC client to the control plane HTTP hostname on port 443. -The Artifact CAS keeps its existing endpoints and native gRPC transport. +The CAS shared endpoint mirrors the control plane (spec D-020): one standard HTTP listener dispatches each request, and native gRPC reaches the native transport through an in-process reverse proxy. `cas.ingress` is one catch-all route on the same hostname — one path `/` with an HTTP/2 (gRPC) upstream — so ByteStream uploads and downloads, browser downloads, and status probes share one Service port and one certificate, including mixed requests on one upstream connection. + +Set gRPC transport annotations under `cas.ingress.annotations`, such as `backend-protocol: GRPCS` for TLS to the pod. The upload body-size annotations keep their existing defaults. + +Enable the shared endpoint with a `cas.ingress` block — for example: + +```yaml +cas: + ingress: + enabled: true + hostname: cas.example.com + tls: true + annotations: + cert-manager.io/cluster-issuer: letsencrypt-prod +``` + +With `tls: true`, configure a certificate source — the cert-manager issuer annotation above, or a pre-created Secret named `cas.example.com-tls` — so the rendered Ingress references a Secret that exists. + +`cas.ingress.enabled` defaults to `false`, so a setup that only copies the documented `cas.ingressAPI` block leaves the shared hostname disabled. Retain `cas.ingressAPI` only while the legacy gRPC endpoint is also required. `cas.serviceAPI`, and the control plane's connection to the legacy CAS endpoint keep their existing settings. CAS adds no gRPC-Web or Connect support. Deployed-controller qualification and the full streaming-performance evidence still need work. See the [experiment instructions](../../app/artifact-cas/README.md#shared-endpoint). The chart configures HTTP/2 upstream traffic for these controllers: @@ -843,7 +861,7 @@ Once done, you can access with [two predefined users](https://github.com/chainlo | `cas.service.loadBalancerIP` | cas service Load Balancer IP | `""` | | `cas.service.loadBalancerSourceRanges` | cas service Load Balancer sources | `[]` | | `cas.service.externalTrafficPolicy` | cas service external traffic policy | `Cluster` | -| `cas.service.annotations` | Additional custom annotations for cas service | `{}` | +| `cas.service.annotations` | Annotations for the shared HTTP and gRPC service. Traefik uses h2c for HTTP/2 upstream traffic. | See values.yaml | | `cas.service.extraPorts` | Extra ports to expose in cas service (normally used with the `sidecars` value) | `[]` | | `cas.service.sessionAffinity` | Control where client requests go, to the same pod or round-robin | `None` | | `cas.service.sessionAffinityConfig` | Additional settings for the sessionAffinity | `{}` | @@ -865,7 +883,7 @@ Once done, you can access with [two predefined users](https://github.com/chainlo | `cas.ingress.hostname` | Default host for the ingress record | `cas.dev.local` | | `cas.ingress.ingressClassName` | IngressClass that will be be used to implement the Ingress (Kubernetes 1.18+) | `""` | | `cas.ingress.path` | Default path for the ingress record | `/` | -| `cas.ingress.annotations` | Additional annotations for the Ingress resource. To enable certificate autogeneration, place here your cert-manager annotations. | `{}` | +| `cas.ingress.annotations` | Annotations for the shared CAS ingress: HTTP/2 (gRPC) upstream plus the upload body limits. | See values.yaml | | `cas.ingress.tls` | Enable TLS configuration for the host defined at `controlplane.ingress.hostname` parameter | `false` | | `cas.ingress.selfSigned` | Create a TLS secret for this ingress record using self-signed certificates generated by Helm | `false` | | `cas.ingress.extraHosts` | An array with additional hostname(s) to be covered with the ingress record | `[]` | diff --git a/deployment/chainloop/chart_test.go b/deployment/chainloop/chart_test.go index 149b81484..50f0f45c5 100644 --- a/deployment/chainloop/chart_test.go +++ b/deployment/chainloop/chart_test.go @@ -28,20 +28,67 @@ import ( "gopkg.in/yaml.v3" ) -// Verify the rendered resources, including templated F5 service names and -// overrides. The legacy gRPC and CAS resources keep their own upstreams. -func TestSharedControlplaneIngress(t *testing.T) { +const ( + casIngressEnabled = "cas.ingress.enabled=true" + helmSet = "--set" + helmSetString = "--set-string" + annotationTrue = "true" +) + +// The control plane serves every protocol per request on its shared ingress +// (replacement spec D-001/D-002). The CAS does the same (spec D-020): the +// shared listener dispatches each request and reaches the native transport +// through an in-process reverse proxy, so one catch-all location with the +// gRPC backend protocol serves every request. The legacy gRPC resources keep +// their own upstreams. +func TestSharedIngress(t *testing.T) { + t.Run("controlplane", func(t *testing.T) { testSharedControlplaneIngress(t) }) + t.Run("cas", func(t *testing.T) { testSharedCASIngress(t) }) +} + +func chartTemplate(t *testing.T, args []string) ([]byte, error) { + t.Helper() + helm, err := exec.LookPath("helm") + if err != nil { + t.Skip("helm is required for chart rendering tests") + } + full := append([]string{"template", "test", ".", helmSet, "development=true", + helmSet, "controlplane.ingressAPI.enabled=true", helmSet, "cas.ingressAPI.enabled=true"}, args...) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, helm, full...) + // Local plugins are unrelated to chart rendering and can print + // startup warnings on stdout before the YAML document. + cmd.Env = append(os.Environ(), "HELM_PLUGINS="+t.TempDir()) + return cmd.CombinedOutput() +} + +func renderChart(t *testing.T, args []string) map[string]chartResource { + t.Helper() + out, err := chartTemplate(t, args) + require.NoError(t, err, "%s", out) + docs := yaml.NewDecoder(bytes.NewReader(out)) + resources := make(map[string]chartResource) + for { + var resource chartResource + err := docs.Decode(&resource) + if errors.Is(err, io.EOF) { + break + } + require.NoError(t, err) + resources[resource.Kind+"/"+resource.Metadata.Name] = resource + } + return resources +} + +// The control plane dispatches each request on one shared HTTP ingress, with +// the gRPC upstream protocol applied to the whole ingress. +func testSharedControlplaneIngress(t *testing.T) { const ( - setFlag = "--set" - setStringFlag = "--set-string" grpcBackend = "GRPC" h2cScheme = "h2c" chartFullname = "test-chainloop" ) - helm, err := exec.LookPath("helm") - if err != nil { - t.Skip("helm is required for chart rendering tests") - } for _, tc := range []struct { name string args []string @@ -50,52 +97,31 @@ func TestSharedControlplaneIngress(t *testing.T) { traefikScheme string }{ {name: "defaults", fullname: chartFullname, backendProtocol: grpcBackend, traefikScheme: h2cScheme}, - {name: "custom name", args: []string{setFlag, "fullnameOverride=shared-api"}, fullname: "shared-api", backendProtocol: grpcBackend, traefikScheme: h2cScheme}, + {name: "custom name", args: []string{helmSet, "fullnameOverride=shared-api"}, fullname: "shared-api", backendProtocol: grpcBackend, traefikScheme: h2cScheme}, { - name: "common annotations merge", - args: []string{setStringFlag, `commonAnnotations.nginx\.org/http2=false`}, - fullname: chartFullname, backendProtocol: grpcBackend, traefikScheme: h2cScheme, + name: "common annotations merge", + args: []string{helmSetString, `commonAnnotations.nginx\.org/http2=false`}, + fullname: chartFullname, + backendProtocol: grpcBackend, traefikScheme: h2cScheme, }, { name: "encrypted upstream overrides", args: []string{ - setStringFlag, `controlplane.ingress.annotations.nginx\.ingress\.kubernetes\.io/backend-protocol=GRPCS`, - setStringFlag, `controlplane.service.annotations.traefik\.ingress\.kubernetes\.io/service\.serversscheme=https`, + helmSetString, `controlplane.ingress.annotations.nginx\.ingress\.kubernetes\.io/backend-protocol=GRPCS`, + helmSetString, `controlplane.service.annotations.traefik\.ingress\.kubernetes\.io/service\.serversscheme=https`, }, fullname: chartFullname, backendProtocol: "GRPCS", traefikScheme: "https", }, } { t.Run(tc.name, func(t *testing.T) { serviceName := tc.fullname + "-controlplane" - args := append([]string{"template", "test", ".", setFlag, "development=true", - setFlag, "controlplane.ingress.enabled=true", setFlag, "controlplane.ingress.hostname=shared.example.test", - setFlag, "controlplane.ingressAPI.enabled=true", setFlag, "cas.ingressAPI.enabled=true"}, tc.args...) - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - cmd := exec.CommandContext(ctx, helm, args...) - // Local plugins are unrelated to chart rendering and can print - // startup warnings on stdout before the YAML document. - cmd.Env = append(os.Environ(), "HELM_PLUGINS="+t.TempDir()) - var stderr bytes.Buffer - cmd.Stderr = &stderr - out, err := cmd.Output() - require.NoError(t, err, "%s", stderr.String()) - docs := yaml.NewDecoder(bytes.NewReader(out)) - resources := make(map[string]chartResource) - for { - var resource chartResource - err := docs.Decode(&resource) - if errors.Is(err, io.EOF) { - break - } - require.NoError(t, err) - resources[resource.Kind+"/"+resource.Metadata.Name] = resource - } + resources := renderChart(t, append(tc.args, + helmSet, "controlplane.ingress.enabled=true", helmSet, "controlplane.ingress.hostname=shared.example.test")) ingress, ok := resources["Ingress/"+serviceName] require.True(t, ok) require.Equal(t, tc.backendProtocol, ingress.Metadata.Annotations["nginx.ingress.kubernetes.io/backend-protocol"]) require.Equal(t, serviceName, ingress.Metadata.Annotations["nginx.org/grpc-services"]) - require.Equal(t, "true", ingress.Metadata.Annotations["nginx.org/http2"]) + require.Equal(t, annotationTrue, ingress.Metadata.Annotations["nginx.org/http2"]) require.Len(t, ingress.Spec.Rules, 1) require.Equal(t, "shared.example.test", ingress.Spec.Rules[0].Host) require.Len(t, ingress.Spec.Rules[0].HTTP.Paths, 1) @@ -109,11 +135,78 @@ func TestSharedControlplaneIngress(t *testing.T) { require.Equal(t, grpcBackend, legacy.Metadata.Annotations["nginx.ingress.kubernetes.io/backend-protocol"]) require.Len(t, legacy.Spec.Rules, 1) require.Equal(t, serviceName+"-api", legacy.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name) - casName := tc.fullname + "-cas-api" - cas := resources["Ingress/"+casName] - require.Equal(t, grpcBackend, cas.Metadata.Annotations["nginx.ingress.kubernetes.io/backend-protocol"]) - require.Len(t, cas.Spec.Rules, 1) - require.Equal(t, casName, cas.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name) + }) + } +} + +// The CAS shared ingress mirrors the control plane (spec D-020): one +// catch-all location, the gRPC backend protocol for its HTTP/2 upstream, and +// the upload body limits. One Service port; no RPC path rules. +func testSharedCASIngress(t *testing.T) { + const ( + grpcBackend = "GRPC" + h2cScheme = "h2c" + chartFullname = "test-chainloop" + ) + for _, tc := range []struct { + name string + args []string + fullname string + backendProtocol string + traefikScheme string + }{ + {name: "defaults", fullname: chartFullname, backendProtocol: grpcBackend, traefikScheme: h2cScheme}, + {name: "custom name", args: []string{helmSet, "fullnameOverride=shared-api"}, fullname: "shared-api", backendProtocol: grpcBackend, traefikScheme: h2cScheme}, + { + name: "common annotations merge", + args: []string{helmSetString, `commonAnnotations.nginx\.org/http2=false`}, + fullname: chartFullname, + backendProtocol: grpcBackend, traefikScheme: h2cScheme, + }, + { + name: "encrypted upstream overrides", + args: []string{ + helmSetString, `cas.ingress.annotations.nginx\.ingress\.kubernetes\.io/backend-protocol=GRPCS`, + helmSetString, `cas.service.annotations.traefik\.ingress\.kubernetes\.io/service\.serversscheme=https`, + }, + fullname: chartFullname, backendProtocol: "GRPCS", traefikScheme: "https", + }, + } { + t.Run(tc.name, func(t *testing.T) { + serviceName := tc.fullname + "-cas" + resources := renderChart(t, append(tc.args, + helmSet, casIngressEnabled, helmSet, "cas.ingress.hostname=shared.example.test")) + + // One catch-all ingress serves every protocol on the shared + // listener, over an HTTP/2 upstream. + ingress, ok := resources["Ingress/"+serviceName] + require.True(t, ok) + require.Equal(t, tc.backendProtocol, ingress.Metadata.Annotations["nginx.ingress.kubernetes.io/backend-protocol"]) + require.Equal(t, serviceName, ingress.Metadata.Annotations["nginx.org/grpc-services"]) + require.Equal(t, annotationTrue, ingress.Metadata.Annotations["nginx.org/http2"]) + require.Equal(t, "100m", ingress.Metadata.Annotations["nginx.ingress.kubernetes.io/proxy-body-size"]) + require.Equal(t, "3M", ingress.Metadata.Annotations["nginx.ingress.kubernetes.io/client-body-buffer-size"]) + require.Len(t, ingress.Spec.Rules, 1) + require.Equal(t, "shared.example.test", ingress.Spec.Rules[0].Host) + require.Len(t, ingress.Spec.Rules[0].HTTP.Paths, 1) + path := ingress.Spec.Rules[0].HTTP.Paths[0] + require.Equal(t, "/", path.Path) + require.Equal(t, serviceName, path.Backend.Service.Name) + require.Equal(t, "http", path.Backend.Service.Port.Name) + + // No second gRPC-path Ingress and no Service port alias remain. + _, ok = resources["Ingress/"+serviceName+"-grpc"] + require.False(t, ok) + svc := resources["Service/"+serviceName] + require.Len(t, svc.Spec.Ports, 1) + require.Equal(t, "http", svc.Spec.Ports[0].Name) + require.Equal(t, tc.traefikScheme, svc.Metadata.Annotations["traefik.ingress.kubernetes.io/service.serversscheme"]) + + // The legacy gRPC ingress keeps its own upstream. + casLegacy := resources["Ingress/"+tc.fullname+"-cas-api"] + require.Equal(t, grpcBackend, casLegacy.Metadata.Annotations["nginx.ingress.kubernetes.io/backend-protocol"]) + require.Len(t, casLegacy.Spec.Rules, 1) + require.Equal(t, tc.fullname+"-cas-api", casLegacy.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name) }) } } @@ -125,6 +218,15 @@ type chartResource struct { Annotations map[string]string `yaml:"annotations"` } `yaml:"metadata"` Spec struct { + Ports []struct { + Name string `yaml:"name"` + Port int `yaml:"port"` + TargetPort string `yaml:"targetPort"` + } `yaml:"ports"` + TLS []struct { + Hosts []string `yaml:"hosts"` + SecretName string `yaml:"secretName"` + } `yaml:"tls"` Rules []struct { Host string `yaml:"host"` HTTP struct { @@ -143,3 +245,13 @@ type chartResource struct { } `yaml:"rules"` } `yaml:"spec"` } + +func TestCASSharedIngressTLS(t *testing.T) { + resources := renderChart(t, []string{helmSet, casIngressEnabled, helmSet, "cas.ingress.tls=true", + helmSetString, `cas.ingress.annotations.cert-manager\.io/cluster-issuer=test-issuer`}) + ingress := resources["Ingress/test-chainloop-cas"] + require.Equal(t, "test-issuer", ingress.Metadata.Annotations["cert-manager.io/cluster-issuer"]) + require.Len(t, ingress.Spec.TLS, 1) + require.Equal(t, "cas.dev.local-tls", ingress.Spec.TLS[0].SecretName) + require.Equal(t, []string{"cas.dev.local"}, ingress.Spec.TLS[0].Hosts) +} diff --git a/deployment/chainloop/templates/cas/ingress.yaml b/deployment/chainloop/templates/cas/ingress.yaml index 3506303dc..beb11b7e9 100644 --- a/deployment/chainloop/templates/cas/ingress.yaml +++ b/deployment/chainloop/templates/cas/ingress.yaml @@ -12,13 +12,8 @@ metadata: namespace: {{ include "common.names.namespace" . | quote }} labels: {{- include "chainloop.cas.labels" . | nindent 4 }} {{- if or .Values.cas.ingress.annotations .Values.commonAnnotations }} - annotations: - {{- if .Values.cas.ingress.annotations }} - {{- include "common.tplvalues.render" ( dict "value" .Values.cas.ingress.annotations "context" $) | nindent 4 }} - {{- end }} - {{- if .Values.commonAnnotations }} - {{- include "common.tplvalues.render" ( dict "value" .Values.commonAnnotations "context" $ ) | nindent 4 }} - {{- end }} + {{- $annotations := include "common.tplvalues.merge" (dict "values" (list .Values.cas.ingress.annotations .Values.commonAnnotations) "context" .) }} + annotations: {{- include "common.tplvalues.render" (dict "value" $annotations "context" $) | nindent 4 }} {{- end }} spec: {{- if and .Values.cas.ingress.ingressClassName (eq "true" (include "common.ingress.supportsIngressClassname" .)) }} @@ -51,9 +46,9 @@ spec: {{- if .Values.cas.ingress.extraRules }} {{- include "common.tplvalues.render" (dict "value" .Values.cas.ingress.extraRules "context" $) | nindent 4 }} {{- end }} - {{- if or (and .Values.cas.ingress.tls (or (include "common.ingress.certManagerRequest" ( dict "annotations" .Values.cas.ingress.annotations )) .Values.cas.ingress.selfSigned)) .Values.cas.ingress.extraTls }} + {{- if or .Values.cas.ingress.tls .Values.cas.ingress.extraTls }} tls: - {{- if and .Values.cas.ingress.tls (or (include "common.ingress.certManagerRequest" ( dict "annotations" .Values.cas.ingress.annotations )) .Values.cas.ingress.selfSigned) }} + {{- if .Values.cas.ingress.tls }} - hosts: - {{ .Values.cas.ingress.hostname | quote }} secretName: {{ printf "%s-tls" .Values.cas.ingress.hostname }} diff --git a/deployment/chainloop/templates/cas/service-http.yaml b/deployment/chainloop/templates/cas/service-http.yaml index 637ead5d4..8fad843f3 100644 --- a/deployment/chainloop/templates/cas/service-http.yaml +++ b/deployment/chainloop/templates/cas/service-http.yaml @@ -37,9 +37,7 @@ spec: - name: http {{- $port := coalesce .Values.cas.service.port .Values.cas.service.ports.http }} port: {{ $port }} - {{- if not (eq $port .Values.cas.containerPorts.http) }} - targetPort: {{ .Values.cas.containerPorts.http }} - {{- end }} + targetPort: http protocol: TCP {{- if and (or (eq .Values.cas.service.type "NodePort") (eq .Values.cas.service.type "LoadBalancer")) (not (empty .Values.cas.service.nodePorts.http)) }} nodePort: {{ .Values.cas.service.nodePorts.http }} diff --git a/deployment/chainloop/values.yaml b/deployment/chainloop/values.yaml index 4fb5d843a..39a3f16ad 100644 --- a/deployment/chainloop/values.yaml +++ b/deployment/chainloop/values.yaml @@ -1129,9 +1129,11 @@ cas: ## ref http://kubernetes.io/docs/tasks/access-application-cluster/create-external-load-balancer/#preserving-the-client-source-ip ## externalTrafficPolicy: Cluster - ## @param cas.service.annotations Additional custom annotations for cas service + ## @extra cas.service.annotations Annotations for the shared HTTP and gRPC service. Traefik uses h2c for HTTP/2 upstream traffic. ## - annotations: {} + annotations: + ## @skip cas.service.annotations.traefik.ingress.kubernetes.io/service.serversscheme + traefik.ingress.kubernetes.io/service.serversscheme: h2c ## @param cas.service.extraPorts Extra ports to expose in cas service (normally used with the `sidecars` value) ## extraPorts: [] @@ -1235,15 +1237,34 @@ cas: ## NOTE: You may need to set this to '/*' in order to use this with ALB ingress controllers ## path: / - ## @param cas.ingress.annotations Additional annotations for the Ingress resource. To enable certificate autogeneration, place here your cert-manager annotations. - ## Use this parameter to set the required annotations for cert-manager, see + ## @extra cas.ingress.annotations Annotations for the shared HTTP, download and native gRPC ingress (spec D-020). + ## The CAS dispatches each request on one listener; native gRPC reaches the + ## native transport through an in-process reverse proxy, so one catch-all + ## location with an HTTP/2 upstream serves every protocol, as on the + ## control plane. Use this parameter to set the required annotations for + ## cert-manager, see ## ref: https://cert-manager.io/docs/usage/ingress/#supported-annotations ## e.g: ## annotations: ## kubernetes.io/cas.ingress.class: nginx ## cert-manager.io/cluster-issuer: cluster-issuer-name ## - annotations: {} + annotations: + ## @skip cas.ingress.annotations.nginx.ingress.kubernetes.io/backend-protocol + ## Tell nginx Ingress Controller to expect gRPC traffic + nginx.ingress.kubernetes.io/backend-protocol: "GRPC" + ## @skip cas.ingress.annotations.nginx.org/grpc-services + nginx.org/grpc-services: '{{ include "chainloop.cas.fullname" . }}' + ## @skip cas.ingress.annotations.nginx.org/http2 + nginx.org/http2: "true" + ## @skip cas.ingress.annotations.nginx.ingress.kubernetes.io/proxy-body-size + # Same upload limit as the independent legacy gRPC ingress. Set 0 to disable. + # Even though we send data in chunks of 1MB, this size refers to all the data sent during the whole streaming session + nginx.ingress.kubernetes.io/proxy-body-size: "100m" + ## @skip cas.ingress.annotations.nginx.ingress.kubernetes.io/client-body-buffer-size + # Improve upload speed by adding client buffering used by http2 control-flows + # https://github.com/chainloop-dev/chainloop/issues/375 + nginx.ingress.kubernetes.io/client-body-buffer-size: "3M" ## @param cas.ingress.tls Enable TLS configuration for the host defined at `controlplane.ingress.hostname` parameter ## TLS certificates will be retrieved from a TLS secret with name: `{{- printf "%s-tls" .Values.controlplane.ingress.hostname }}` ## You can: diff --git a/docs/specs/issue-3628-single-endpoint.md b/docs/specs/issue-3628-single-endpoint.md index b23215c7c..e35f61717 100644 --- a/docs/specs/issue-3628-single-endpoint.md +++ b/docs/specs/issue-3628-single-endpoint.md @@ -26,7 +26,8 @@ Control-plane requirements and design superseded by [Spec issue-3628-controlplan For the control plane, the requirements, constraints, proposal, diagram, decision records, milestones, and risks below are historical. Use the replacement spec for all control-plane implementation and qualification. Only CAS requirements remain in scope here. -CAS transport selection and qualification remain open under the parent ticket. +The CAS transport decision is recorded below in D-020. +CAS qualification remains open under the parent ticket. ## Summary @@ -56,14 +57,29 @@ The control plane and the Artifact CAS will each serve gRPC and HTTP on one port ## Requirements -### R-001: One port serves all protocols -On its single port, each service MUST serve gRPC (HTTP/2) and plain HTTP (HTTP/1.1 and HTTP/2). The control plane MUST also serve gRPC-Web. The CAS has no gRPC-Web today and does not get it (see D-011). This includes every endpoint that the gRPC port and the HTTP port serve today. -- Done when: these clients work against only the single port: - - the CLI and the web UI - - the OIDC login and the status probes - - the referrer discovery and the OpenAPI document - - CAS browser downloads -- Done when: the same checks pass through nginx ingress and through Traefik ingress. +### ~~R-001: One port serves all protocols~~ +Superseded by R-009 for the CAS. The control-plane scope moved to [the replacement spec](issue-3628-controlplane-http-dispatch.md). + +~~On its single port, each service MUST serve gRPC (HTTP/2) and plain HTTP (HTTP/1.1 and HTTP/2). The control plane MUST also serve gRPC-Web. The CAS has no gRPC-Web today and does not get it (see D-011). This includes every endpoint that the gRPC port and the HTTP port serve today.~~ +~~- ~~Done when: these clients work against only the single port:~~~~ +~~ - ~~the CLI and the web UI~~~~ +~~ - ~~the OIDC login and the status probes~~~~ +~~ - ~~the referrer discovery and the OpenAPI document~~~~ +~~ - ~~CAS browser downloads~~~~ +~~- ~~Done when: the same checks pass through nginx ingress and through Traefik ingress.~~~~ + +### ~~R-009: CAS shared endpoint protocol scope~~ +Superseded by R-010. +~~Supersedes R-001 for the CAS. On its single port, the CAS MUST serve native gRPC over HTTP/2 and plain HTTP over HTTP/1.1, through cmux connection classification with the native gRPC transport (D-017). Plain HTTP/2 other than native gRPC is not supported on the CAS shared port: the gRPC matcher's SETTINGS injection breaks such a connection once it falls through to the HTTP server. Plain HTTP/2 and mixed connections remain control-plane scope under the replacement spec.~~ +~~- Done when: focused tests cover native gRPC over h2c and TLS, HTTP/1.1 downloads, the status endpoints, and the CORS preflight through the shared port.~~ +~~- Done when: the chart routes the gRPC paths and the HTTP paths into separate upstream pools, with HTTP/1.1 for the HTTP routes (D-018/D-019).~~ +~~- Done when: browser downloads through a configured ingress work, and the documentation states that a browser downloading directly from a TLS-configured shared port is not a supported deployment.~~ + +### R-010: CAS shared endpoint protocol scope +Supersedes R-009. On its single port, the CAS MUST serve native gRPC over HTTP/2 and plain HTTP over HTTP/1.1 and HTTP/2, dispatched per request (D-020), including mixed native-gRPC and plain-HTTP requests on one upstream HTTP/2 connection in any order. ByteStream MUST reach the native transport through the in-process reverse proxy. +- Done when: focused tests cover native gRPC over h2c and TLS, HTTP/1.1 and HTTP/2 requests, the status endpoints, and the CORS preflight through the shared port, and the same checks pass with mixed request types on one pooled connection. +- Done when: the 1 GiB transport comparison through the shared endpoint stays within 10% of the legacy native listener with memory bounded and not growing with artifact size. +- Done when: browser downloads through a configured ingress work. ### R-005: Existing configuration keeps working An installation that does not change its configuration MUST keep working during the deprecation period. The legacy gRPC port keeps serving gRPC at its current address during that period. CLI configurations that use ingress hostnames keep working when the ingress backend moves to the single port. The control plane connection to the CAS, and any other client that uses the legacy gRPC address, MUST move to the single HTTP address before the legacy gRPC port is disabled. @@ -159,11 +175,11 @@ The current HTTP address becomes the single port. It gets its own TLS settings, Superseded by D-013. ~~The service splits by connection in front of the current gRPC and HTTP servers.~~ -- ~~Why: it keeps the whole middleware chain and the native gRPC transport for CAS streams, and it needs the smallest change.~~ -- ~~Alternatives considered:~~ - - ~~One HTTP handler that routes each request: gRPC calls then use the HTTP adapter of the gRPC server. The gRPC library documents that path as slower and with fewer features. That puts R-003 at risk.~~ - - ~~A move to Connect: it replaces every middleware and route, which is a large migration for the same result.~~ -- ~~Source: drafting~~ +~~- ~~Why: it keeps the whole middleware chain and the native gRPC transport for CAS streams, and it needs the smallest change.~~~~ +~~- ~~Alternatives considered:~~~~ +~~ - ~~One HTTP handler that routes each request: gRPC calls then use the HTTP adapter of the gRPC server. The gRPC library documents that path as slower and with fewer features. That puts R-003 at risk.~~~~ +~~ - ~~A move to Connect: it replaces every middleware and route, which is a large migration for the same result.~~~~ +~~- ~~Source: drafting~~~~ ### D-003: Deprecation default The legacy gRPC port is on by default, then off by default, then deleted. @@ -184,10 +200,10 @@ The chart uses one Service and one hostname per service. Only where the controll Superseded by D-015. ~~The services solve it in the server.~~ -- ~~Why: it works with any ingress, any proxy, and outside Kubernetes.~~ -- ~~Alternatives considered:~~ - - ~~A chart-level proxy or Gateway API routes (#2642): they need extra infrastructure and leave two ports in the service.~~ -- ~~Source: drafting~~ +~~- ~~Why: it works with any ingress, any proxy, and outside Kubernetes.~~~~ +~~- ~~Alternatives considered:~~~~ +~~ - ~~A chart-level proxy or Gateway API routes (#2642): they need extra infrastructure and leave two ports in the service.~~~~ +~~- ~~Source: drafting~~~~ ### D-006: Metrics and profiler ports The metrics and profiler ports stay separate. @@ -219,10 +235,10 @@ The Helm chart change stays in this spec, as milestone 2. Superseded by D-014. ~~The single port routes each connection once. It supports only clients and proxies that keep gRPC and plain HTTP on separate HTTP/2 connections.~~ -- ~~Why: a connection that the native gRPC server owns cannot pass single requests to the HTTP server. Native gRPC transport for CAS streams (R-003) needs the connection split of D-001.~~ -- ~~Alternatives considered:~~ - - ~~Route each request on one HTTP/2 server: it supports every proxy. But all gRPC calls then use the slower HTTP adapter of the gRPC server, which puts R-003 at risk.~~ -- ~~Source: [cubic review](https://github.com/chainloop-dev/chainloop/pull/3629#discussion_r4286475927)~~ +~~- ~~Why: a connection that the native gRPC server owns cannot pass single requests to the HTTP server. Native gRPC transport for CAS streams (R-003) needs the connection split of D-001.~~~~ +~~- ~~Alternatives considered:~~~~ +~~ - ~~Route each request on one HTTP/2 server: it supports every proxy. But all gRPC calls then use the slower HTTP adapter of the gRPC server, which puts R-003 at risk.~~~~ +~~- ~~Source: [cubic review](https://github.com/chainloop-dev/chainloop/pull/3629#discussion_r4286475927)~~~~ ### D-011: gRPC-Web only on the control plane The single port serves gRPC-Web on the control plane only. The CAS serves gRPC and HTTP. @@ -246,13 +262,13 @@ Drops D-002. The current HTTP port becomes the single port, not the gRPC port. T Superseded by D-016. ~~Supersedes D-001. The service splits by connection in front of the current gRPC and HTTP servers, with the cmux library as the connection classifier. The library owns the connection buffering and the listener plumbing. The service owns only a small matcher and the HTTP request dispatcher. The matcher sends the server's HTTP/2 settings frame to clients that wait for it. It reads the first request's content type through the standard HTTP/2 library. The matcher closes the connection when its first request is not native gRPC. The dispatcher carries the gRPC-Web wrapper and the native-gRPC fallback.~~ -- ~~Why: it keeps the whole middleware chain and the native gRPC transport for CAS streams. It also takes HTTP/2 parsing out of the code this project maintains. Argo CD and etcd run the same library split in production.~~ -- ~~Why: the gRPC library documents that its HTTP adapter may differ in performance and features from the native transport. It does not document the adapter as slower, and this project has not measured a difference. The native transport stays the normal path for gRPC connections, so R-003 does not rest on the adapter.~~ -- ~~Alternatives considered:~~ - - ~~The custom connection parser of the first implementation: it works, but the project then maintains its own HTTP/2 frame parsing, replay, and resource bounds.~~ - - ~~One HTTP handler that routes each request: every gRPC call, on every connection, then uses the HTTP adapter of the gRPC server.~~ - - ~~A move to Connect: it requires migrating the generated server interfaces and the interceptors. Ordinary HTTP routes can stay, but the migration is larger than the classifier swap for the same single port.~~ -- ~~Source: [PR comment](https://github.com/chainloop-dev/chainloop/pull/3647#issuecomment-6266145446)~~ +~~- ~~Why: it keeps the whole middleware chain and the native gRPC transport for CAS streams. It also takes HTTP/2 parsing out of the code this project maintains. Argo CD and etcd run the same library split in production.~~~~ +~~- ~~Why: the gRPC library documents that its HTTP adapter may differ in performance and features from the native transport. It does not document the adapter as slower, and this project has not measured a difference. The native transport stays the normal path for gRPC connections, so R-003 does not rest on the adapter.~~~~ +~~- ~~Alternatives considered:~~~~ +~~ - ~~The custom connection parser of the first implementation: it works, but the project then maintains its own HTTP/2 frame parsing, replay, and resource bounds.~~~~ +~~ - ~~One HTTP handler that routes each request: every gRPC call, on every connection, then uses the HTTP adapter of the gRPC server.~~~~ +~~ - ~~A move to Connect: it requires migrating the generated server interfaces and the interceptors. Ordinary HTTP routes can stay, but the migration is larger than the classifier swap for the same single port.~~~~ +~~- ~~Source: [PR comment](https://github.com/chainloop-dev/chainloop/pull/3647#issuecomment-6266145446)~~~~ ### D-014: Separate upstream connections, one-way fallback Supersedes D-010. The single port routes each connection once. It supports clients and proxies that keep native gRPC and plain HTTP on separate upstream HTTP/2 connections. Supported ingress configurations use separate upstream pools for the two kinds of traffic. The chart documents and tests them for nginx and Traefik. A connection that the HTTP server owns may carry a native gRPC request later. The HTTP server then gives the request to the gRPC server through the documented fallback, on the adapter transport. The reverse is not supported: a connection that the native gRPC server owns cannot carry a plain HTTP request. Arbitrary mixed native-gRPC and plain-HTTP requests on one shared upstream connection are not required. @@ -269,18 +285,68 @@ Supersedes D-005. The services solve the multiplexing in the server process. The - A mandatory sidecar or embedded proxy: it works behind more proxy configurations, but it adds a process and its lifecycle to every deployment. - Source: [PR comment](https://github.com/chainloop-dev/chainloop/pull/3647#issuecomment-6266145446) -### D-016: Complete the classifier's SETTINGS exchange before HTTP handoff -Supersedes D-013. The service keeps cmux connection classification, the native gRPC server, and the HTTP request dispatcher with its gRPC-Web wrapper and native-gRPC fallback. Both native gRPC and ordinary HTTP/2 clients may wait for server SETTINGS before their first request. Silence does not identify the protocol of that request. -- The matcher uses the standard HTTP/2 framer and HPACK decoder. It limits incoming frames to the initial 16 KiB maximum before reading their payloads, as well as limiting classification bytes, time, and concurrency. -- When the matcher sends an empty SETTINGS frame and the first request is HTTP, it waits within the classification bounds for that SETTINGS acknowledgement. It records the validated ACK's exact offset in cmux's replay. The HTTP listener removes only that nine-byte frame before the HTTP server reads the connection. The HTTP server owns all subsequent SETTINGS exchanges. An incomplete injected handshake is closed. -- Native gRPC connections still go to the existing native server with their original replay. HTTP connections that did not need an injected SETTINGS frame also retain their original replay. HTTP requests and later streams are not parsed again by the handoff adapter. -- Why: rejecting every non-gRPC request after the matcher sends SETTINGS also rejects valid HTTP-only clients. Forwarding the extra ACK unchanged makes net/http reject the connection after the client acknowledges the HTTP server's own SETTINGS. Removing the one ACK preserves its ownership without adding an ongoing frame parser. -- Evidence: focused tests cover HTTP and gRPC-Web content types over plaintext and TLS, ACKs before and after the first request, subsequent requests on the same connection, incomplete handshakes, and an oversized frame header without its payload. These tests do not establish full R-002 parity or R-003 CAS performance. -- Source: [implementation review](https://github.com/chainloop-dev/chainloop/pull/3647#pullrequestreview-5573391966). +### ~~D-016: Complete the classifier's SETTINGS exchange before HTTP handoff~~ +Superseded by D-017. + +~~Supersedes D-013. The service keeps cmux connection classification, the native gRPC server, and the HTTP request dispatcher with its gRPC-Web wrapper and native-gRPC fallback. Both native gRPC and ordinary HTTP/2 clients may wait for server SETTINGS before their first request. Silence does not identify the protocol of that request.~~ +~~- ~~The matcher uses the standard HTTP/2 framer and HPACK decoder. It limits incoming frames to the initial 16 KiB maximum before reading their payloads, as well as limiting classification bytes, time, and concurrency.~~~~ +~~- ~~When the matcher sends an empty SETTINGS frame and the first request is HTTP, it waits within the classification bounds for that SETTINGS acknowledgement. It records the validated ACK's exact offset in cmux's replay. The HTTP listener removes only that nine-byte frame before the HTTP server reads the connection. The HTTP server owns all subsequent SETTINGS exchanges. An incomplete injected handshake is closed.~~~~ +~~- ~~Native gRPC connections still go to the existing native server with their original replay. HTTP connections that did not need an injected SETTINGS frame also retain their original replay. HTTP requests and later streams are not parsed again by the handoff adapter.~~~~ +~~- ~~Why: rejecting every non-gRPC request after the matcher sends SETTINGS also rejects valid HTTP-only clients. Forwarding the extra ACK unchanged makes net/http reject the connection after the client acknowledges the HTTP server's own SETTINGS. Removing the one ACK preserves its ownership without adding an ongoing frame parser.~~~~ +~~- ~~Evidence: focused tests cover HTTP and gRPC-Web content types over plaintext and TLS, ACKs before and after the first request, subsequent requests on the same connection, incomplete handshakes, and an oversized frame header without its payload. These tests do not establish full R-002 parity or R-003 CAS performance.~~~~ +~~- ~~Source: [implementation review](https://github.com/chainloop-dev/chainloop/pull/3647#pullrequestreview-5573391966).~~~~ + +### ~~D-017: CAS shared endpoint: cmux with the native transport, and the corrected SETTINGS contract~~ +Superseded by D-020. +~~Supersedes D-016 for the CAS endpoint. The control-plane parts of D-014 to D-016 were already superseded by [the replacement spec](issue-3628-controlplane-http-dispatch.md).~~ + +~~The CAS shared endpoint is `server.http.addr`. cmux reads the start of each new connection and hands the whole connection to one of the two existing transports: an HTTP/2 connection whose first request declares a native gRPC content type goes to the existing native gRPC listener (`grpc.Server.Serve`); every other connection goes to the existing HTTP server. ByteStream uploads and downloads therefore keep the grpc-go native transport, so R-003 does not rest on the HTTP adapter.~~ +~~- Why not the merged control-plane design, per-request `ServeHTTP` dispatch: the [PR #3663 1 GiB comparison](https://github.com/chainloop-dev/chainloop/pull/3663) measured about 1.25 GiB of combined client/server sampled peak live heap for adapter uploads, against about 28 MiB for the native transport, and download throughput 11.8% lower, already past the R-003 limit. grpc-go's handler transport reads the HTTP body in a separate goroutine into an unbounded receive buffer, so read-ahead is not bounded by the application's pace. CAS moves gigabyte streams; the control plane does not. The two services keep different transport designs on purpose.~~ +~~- Why D-016 is superseded: D-016 assumed a classifier this project owns, which records the SETTINGS ACK offset in its own replay and removes that frame before the HTTP server reads. The cmux library owns the sniffing buffer, so that removal does not exist. Two reproductions on the CAS branch fix the contract instead. First, without SETTINGS injection, direct grpc-go clients deadlock: they send their preface and SETTINGS and wait for the server's SETTINGS before their first HEADERS frame, while the matcher waits for headers to sniff. The gRPC matcher must use cmux's SETTINGS-writing matcher. Second, that matcher writes a SETTINGS frame on every HTTP/2 connection, matched or not, because every HTTP/2 client sends SETTINGS with its preface. A plain HTTP/2 client that falls through to the HTTP server acknowledges the injected SETTINGS frame, and net/http rejects the unsolicited acknowledgement (GOAWAY, PROTOCOL_ERROR).~~ +~~- Contract: once the matcher has written to a connection, the connection belongs to the native gRPC server. The HTTP side of the shared endpoint supports HTTP/1.1 only; HTTP/2 reaches the CAS only as native gRPC. Supported proxies keep native gRPC and plain HTTP on separate upstream connections (D-014) and dial HTTP/1.1 for the HTTP routes (D-018).~~ +~~- The CAS has no per-request gRPC adapter fallback. D-017 supersedes that part of D-014 for CAS. All supported ByteStream transfers use the native transport.~~ +~~- Classification limits apply before a server owns the connection. They bound time, bytes read from the socket, and concurrent classifications. Shutdown closes pending sockets. These limits end at handoff and do not limit active transfers.~~ +~~- Source: [PR #3663](https://github.com/chainloop-dev/chainloop/pull/3663) experiment and benchmark, and the local reproductions on its branch.~~ + +### ~~D-018: CAS chart: one hostname, path-split upstream pools, HTTP/1.1 for the HTTP routes~~ +Superseded by D-020. +~~The chart exposes one CAS hostname and one public port, and splits the routes by protocol into separate upstream pools: the Google ByteStream, CAS, health, reflection, metadata, and channelz RPC paths go to the gRPC pool with an HTTP/2 upstream (nginx `backend-protocol: GRPC` and `grpc-services`; the upload body-size annotations move with them), and `/download/*`, `/statusz`, and `/infoz` go to the HTTP pool, which dials HTTP/1.1, not h2c. One certificate terminates at the edge.~~ +~~- Why the path split is exact for CAS: CAS has no gRPC-Web (D-011), so its gRPC paths and its HTTP paths are disjoint. No path can carry both protocols. The control plane could not use this split because native gRPC and gRPC-Web share RPC paths; the [PR #3647 NGINX reproduction](https://github.com/chainloop-dev/chainloop/pull/3647#issuecomment-6272007730) showed the path-only split failing there with 415 on gRPC-Web. CAS does not have that failure mode.~~ +~~- Why HTTP/1.1 upstream on the HTTP pool: D-017. An HTTP/2 upstream for the HTTP routes triggers the matcher's SETTINGS injection, and the client's acknowledgement then breaks the HTTP server connection with GOAWAY.~~ +~~- Controller notes: community nginx merges same-host paths from multiple Ingress resources and sets the backend protocol per resource, which expresses the split. F5 NGINX conflicts on unrelated same-host Ingresses; use its VirtualServer resource. Traefik applies the `service.serversscheme` annotation per Service, so one Service cannot be dialed h2c for the gRPC routes and HTTP/1.1 for the HTTP routes; the Traefik split needs a second Service for the HTTP routes or a controller-specific resource. The browser download endpoint stays on the HTTP routes: it is published API ([docs](https://docs.chainloop.dev/api-reference/downloadservice/download-artifacts-from-cas)), and CAS does not drop it.~~ +~~- Source: chart changes on the [PR #3663](https://github.com/chainloop-dev/chainloop/pull/3663) branch. Connection reuse and certificate ownership are specified in D-019.~~ + +### ~~D-019: CAS ingress connection reuse and certificate ownership~~ +Superseded by D-020. +~~Supplements D-018. Ingress-nginx shares a global keepalive cache. Separate Ingress resources alone do not isolate their connections.~~ + +~~The CAS Service has an HTTP port and a gRPC port alias. Both target the same application listener. The HTTP routes use pod endpoints. The gRPC routes use the Service ClusterIP and the alias. Each protocol gets a different backend identity and upstream address. This keeps one Service, one hostname, and one application listener (R-008).~~ + +~~The Service port numbers MUST differ. At least one upstream MUST use the ClusterIP. The chart rejects configurations that use pod endpoints for both protocols, or a headless Service.~~ + +~~Both Ingress resources reference the same TLS Secret when shared TLS is enabled. Only one requests the certificate. The HTTP Ingress owns that request when its annotations configure a certificate issuer. Otherwise, the gRPC Ingress can request it. An existing TLS Secret needs no issuer annotation.~~ + +~~- Why: a single-worker ingress-nginx reproduction returned 502 for gRPC after an HTTP request warmed the shared connection cache. Distinct backend identities and addresses passed the same sequence with authenticated ByteStream and browser downloads.~~ +~~- Compatibility: the legacy CAS listener, Service, Ingress, and their settings retain their existing behavior. The control-plane connection continues to use that legacy endpoint.~~ +~~- Evidence limit: the local controller check covers ingress-nginx. F5 and Traefik need separate qualification. The full R-002 and original R-003 evidence remains outstanding.~~ +~~- Source: [PR #3663 implementation review](https://github.com/chainloop-dev/chainloop/pull/3663#issuecomment-6276983988).~~ + +### D-020: CAS shared endpoint: per-request dispatch with a standard-library reverse proxy to a private native server +Supersedes D-017, D-018, and D-019. The merged control-plane design stays unchanged. + +The CAS shared endpoint is `server.http.addr`, owned by one standard HTTP listener (HTTP/1.1, HTTP/2, optional TLS). Each request is dispatched: native gRPC — HTTP/2 with a content type in the `application/grpc` family, excluding `application/grpc-web*`, which CAS does not serve (D-011) — goes through an in-process `httputil.ReverseProxy` to a private native gRPC server bound to an ephemeral loopback address; every other request uses the existing HTTP routes. The private server runs the same services and interceptors as the legacy listener and is independent of it: the legacy flag, address, and TLS settings are unchanged, and the private listener exists whether or not the legacy listener is enabled. +- Why not cmux (D-017): connection classification cannot serve the single-location all-HTTP/2 upstream that the control-plane chart shape produces. Once the SETTINGS-writing matcher has spoken, the connection is gRPC-only, and pooled mixed connections then fail — the 502 reproduction that D-019's Service port alias worked around. Per-request dispatch removes that contract instead of working around it: mixed native-gRPC and plain-HTTP requests share one upstream connection in any order. +- Why the proxy and not `grpc.Server.ServeHTTP` for gRPC: the measured adapter read-ahead, about 1.25 GiB of live heap on a 1 GiB upload against about 28 MiB on the native transport. `httputil.ReverseProxy` streams the request body into the private connection, whose HTTP/2 flow-control window paces the proxy's reads while the shared listener's own window paces the client, so no unbounded read-ahead exists in either direction. gRPC frames, compression, metadata, and response trailers pass through without decoding or rebuilding. +- Why a private loopback listener and not the legacy listener: the legacy listener carries independent address and TLS settings and is optional. An h2c proxy cannot dial a TLS-configured address, and the shared endpoint must work with the legacy listener disabled. +- Chart: one Service port and one catch-all Ingress on the same hostname, matching the control-plane shape — one path `/` with nginx `backend-protocol: GRPC` (HTTP/2 upstream), `grpc-services`, and the upload body-size annotations, and the Traefik h2c Service annotation. No RPC path rules, no second Ingress, no Service port alias. +- Measured (local, three runs per operation, 1 GiB, disk-backed staging): shared-endpoint median upload 212.89 MB/s against legacy native 233.92 MB/s (−9.0%), and median download 177.11 MB/s against 192.82 MB/s (−8.1%), with sampled peak live heap of about 32–43 MiB against about 29–32 MiB. Within the R-003 10% shape with a thin margin; the release and object-store qualification remains open. +- Evidence limits: mixed-connection, h2c and TLS, authentication/error, cancellation, and shutdown-drain tests pass on the branch, and a standalone probe covered gzip, binary metadata, trailers, rich error details, and downstream cancellation through the standard proxy. The full R-002 matrix, the original R-003 qualification, and F5/Traefik controller qualification remain outstanding. +- The branch carries the production change only. The local kind ingress-nginx harness and the raw benchmark output file were removed at the owner's request; the raw benchmark numbers and the kind reproduction live in the [PR #3663 discussion](https://github.com/chainloop-dev/chainloop/pull/3663), and the benchmark harness and its command stay in `app/artifact-cas`. +- Sources: [proxy review and compatibility probe](https://github.com/chainloop-dev/chainloop/pull/3663#issuecomment-6278243434), and the [PR #3663](https://github.com/chainloop-dev/chainloop/pull/3663) implementation. ## Open Questions -- [x] **Must the single port support proxies that share one upstream connection for gRPC and plain HTTP?** Examples: Envoy-based proxies, service meshes. Answered by D-014: not required. Setups that share one upstream pool can route gRPC to the legacy gRPC port during the deprecation period. The chart documents the supported nginx and Traefik configurations, which keep the two kinds of traffic on separate upstream connections. Source: [cubic review](https://github.com/chainloop-dev/chainloop/pull/3629#discussion_r4286475927). +- [x] **Must the single port support proxies that share one upstream connection for gRPC and plain HTTP?** Examples: Envoy-based proxies, service meshes. Answered by D-014 for the connection-classification designs, which could not accept them. Superseded for the CAS by R-010 and D-020: the per-request dispatcher accepts mixed native-gRPC and plain-HTTP requests on one upstream HTTP/2 connection in any order, so a shared upstream pool works for the CAS shared endpoint. The control-plane replacement spec accepts mixed connections from the start. The legacy gRPC port remains available during the deprecation period. Source: [cubic review](https://github.com/chainloop-dev/chainloop/pull/3629#discussion_r4286475927). D-007 and D-008 record the answers to the questions from drafting.