diff --git a/builder.go b/builder.go index 1f70e77..b3a4efe 100644 --- a/builder.go +++ b/builder.go @@ -37,11 +37,11 @@ import ( // plus the services.Base chain via // *pythonservice.Service. // FastAPI *Service — fastapi-specific state: richer Settings -// and the REST endpoint. +// and the endpoint handles. // // Inherited: Init. -// Overridden: Load (fastapi puts source under ./code, discovers REST -// endpoint), Update (applies builder templates), Sync (gRPC codegen for +// Overridden: Load (fastapi puts source under ./code, finds the HTTP +// endpoint, if any), Update (applies builder templates), Sync (gRPC codegen for // declared dependencies and the optional service-owned gRPC server), Build // (Dockerfile and build plan preparation), // Deploy (k8s), Create (two-question Communicate + REST endpoint). @@ -61,8 +61,9 @@ func NewBuilder(svc *Service) *Builder { } } -// Load overrides generic to place source under /code and to -// discover the REST endpoint. +// Load overrides generic to place source under /code and to find +// the endpoint the app's HTTP listener serves, which a service whose faces are +// gRPC and Connect alone does not declare. func (s *Builder) Load(ctx context.Context, req *builderv0.LoadRequest) (*builderv0.LoadResponse, error) { defer s.Wool.Catch() @@ -71,9 +72,8 @@ func (s *Builder) Load(ctx context.Context, req *builderv0.LoadRequest) (*builde // spec into its own half of the settings alone, leaving grpc-server and // connect-endpoint unset on every loaded service, so Sync generated no // server code and Deploy rendered no gRPC or Connect port. It reports a - // failure in the response's state, not the error: stop there, or the - // endpoint lookup below replaces the real cause with "no rest endpoint - // found". + // failure in the response's state, not the error: stop there, so the + // response carries that cause. resp, err := s.Base.Builder.LoadService(ctx, req, services.BuilderLoad{Settings: s.FastAPI.Settings}) s.Service.SourceLocation = s.Service.ResolveSourceLocation() if err != nil || resp.GetState().GetState() == builderv0.LoadStatus_ERROR { @@ -96,7 +96,7 @@ func (s *Builder) Load(ctx context.Context, req *builderv0.LoadRequest) (*builde return s.Base.Builder.LoadResponse() } - s.FastAPI.RestEndpoint, err = resources.FindRestEndpoint(ctx, s.Endpoints) + s.FastAPI.HTTPEndpoint, err = findHTTPEndpoint(ctx, s.Endpoints) if err != nil { return s.Base.Builder.LoadError(err) } @@ -552,20 +552,27 @@ func preparePodOverlay(overlay *services.PodTemplateOverlay) error { func (s *Builder) Deploy(ctx context.Context, req *builderv0.DeploymentRequest) (*builderv0.DeploymentResponse, error) { defer s.Wool.Catch() + // The rendered pod is probed on the app's HTTP listener, the container's + // port 8080, which a service with no HTTP endpoint does not publish: say so + // rather than report the missing endpoint as missing health. + listener := s.FastAPI.HTTPEndpoint + if listener == nil { + return nil, fmt.Errorf("the deployment probes the HTTP endpoint (api: rest or api: http), and the service declares none: declare the endpoint its probes answer on") + } // The current Core manifest profile requires all three intents. Require // an authored startup/restart policy instead of inventing one from readiness. - plan := resources.PlanEndpointProbes(s.FastAPI.RestEndpoint) + plan := resources.PlanEndpointProbes(listener) if plan.Startup == nil || plan.Liveness == nil { return nil, fmt.Errorf("the Kubernetes manifest profile requires explicit endpoint health.startup and health.liveness; readiness is not a restart policy") } // A restricted render is what a cell runs, and a cell's execution // admission refuses an exec probe: probe natively there. - probes, err := deploymentProbesFor(s.FastAPI.RestEndpoint, req.GetDeployment().GetKubernetes().GetProfile()) + probes, err := deploymentProbesFor(listener, req.GetDeployment().GetKubernetes().GetProfile()) if err != nil { return nil, err } for _, endpoint := range s.Endpoints { - if endpoint.GetName() != s.FastAPI.RestEndpoint.GetName() && endpoint.GetHealth() != nil { + if endpoint.GetName() != listener.GetName() && endpoint.GetHealth() != nil { return nil, fmt.Errorf("declared health on additional endpoint %q requires a combined deployment probe; refusing to ignore it", endpoint.GetName()) } } @@ -665,12 +672,9 @@ func (s *Builder) Create(ctx context.Context, _ *builderv0.CreateRequest) (*buil return s.Base.Builder.CreateResponse(ctx, s.FastAPI.Settings) } -// CreateEndpoints materializes the REST endpoint. -// -// openapi/api.swagger.json is generated at Runtime time by src/openapi.py -// (uv run python src/openapi.py). At Create time it doesn't exist yet — -// that specific error is expected and swallowed with a debug log. Any -// other LoadRestAPI failure (malformed JSON, permission, …) propagates. +// CreateEndpoints materializes the REST endpoint from the OpenAPI document +// the factory templates have just written (openapi/api.swagger.json); Init +// regenerates it with src/openapi.py from then on. func (s *Builder) CreateEndpoints(ctx context.Context) error { openapiFile := s.Local("openapi/api.swagger.json") endpoint := s.Base.BaseEndpoint(standards.REST) @@ -694,13 +698,9 @@ func (s *Builder) CreateEndpoints(ctx context.Context) error { endpoint.Exposure = resources.ExposurePublic } - rest, loadErr := resources.LoadRestAPI(ctx, shared.Pointer(openapiFile)) - if loadErr != nil { - if !errors.Is(loadErr, os.ErrNotExist) && !isFileNotExistErr(loadErr) { - return s.Wool.Wrapf(loadErr, "cannot load rest api") - } - s.Wool.Debug("openapi spec not generated yet (expected at Create time)", - wool.Field("path", openapiFile)) + rest, err := resources.LoadRestAPI(ctx, shared.Pointer(openapiFile)) + if err != nil { + return s.Wool.Wrapf(err, "cannot load rest api") } api, err := resources.NewAPI(ctx, endpoint, resources.ToRestAPI(rest)) @@ -711,8 +711,8 @@ func (s *Builder) CreateEndpoints(ctx context.Context) error { if err != nil { return s.Wool.Wrapf(err, "cannot declare scaffold health") } - s.FastAPI.RestEndpoint = api - s.Endpoints = []*basev0.Endpoint{s.FastAPI.RestEndpoint} + s.FastAPI.HTTPEndpoint = api + s.Endpoints = []*basev0.Endpoint{api} if s.FastAPI.Settings.GRPCServer.Enabled { grpcEndpoint, grpcErr := s.grpcEndpoint(ctx) @@ -755,13 +755,6 @@ func (s *Builder) grpcEndpoint(ctx context.Context) (*basev0.Endpoint, error) { return api, nil } -// isFileNotExistErr matches the bespoke "file does not exist" error string -// that resources.LoadRestAPI returns for missing files — it doesn't wrap -// os.ErrNotExist, so errors.Is can't catch it directly. -func isFileNotExistErr(err error) bool { - return err != nil && err.Error() == "file does not exist" -} - func (s *Builder) populateSettingsFromAnswers() error { var err error if s.FastAPI.Settings.HotReload, err = communicate.Confirm(s.answers, HotReload); err != nil { diff --git a/composed_project_test.go b/composed_project_test.go index 1185516..301a92f 100644 --- a/composed_project_test.go +++ b/composed_project_test.go @@ -60,7 +60,7 @@ func TestContainerEnvironmentCarriesAProjectOutsideTheWorkspace(t *testing.T) { runtime.Service.SourceLocation = source endpoint := &basev0.Endpoint{Module: "mod", Service: runtime.Identity.Name, Name: "rest", Api: standards.REST} - runtime.FastAPI.RestEndpoint = endpoint + runtime.FastAPI.HTTPEndpoint = endpoint runtime.NetworkMappings = []*basev0.NetworkMapping{{ Endpoint: endpoint, Instances: []*basev0.NetworkInstance{{ diff --git a/connect_only_test.go b/connect_only_test.go new file mode 100644 index 0000000..faa60e8 --- /dev/null +++ b/connect_only_test.go @@ -0,0 +1,333 @@ +package main + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + basev0 "github.com/codefly-dev/core/generated/go/codefly/base/v0" + agentv0 "github.com/codefly-dev/core/generated/go/codefly/services/agent/v0" + builderv0 "github.com/codefly-dev/core/generated/go/codefly/services/builder/v0" + runtimev0 "github.com/codefly-dev/core/generated/go/codefly/services/runtime/v0" + "github.com/codefly-dev/core/resources" + "github.com/codefly-dev/core/standards" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "gopkg.in/yaml.v3" +) + +// A Connect-only service declares its gRPC server and the Connect face over it, +// and no HTTP endpoint: no REST face, no OpenAPI document, no probe listener. +// These tests hold the agent to loading, syncing and starting one. + +// TestFindHTTPEndpoint pins which endpoint the app's HTTP listener serves. +func TestFindHTTPEndpoint(t *testing.T) { + ctx := context.Background() + rest := &basev0.Endpoint{Name: "rest", Api: standards.REST} + http := &basev0.Endpoint{Name: "probes", Api: standards.HTTP} + grpc := &basev0.Endpoint{Name: "grpc", Api: standards.GRPC} + connect := &basev0.Endpoint{Name: "connect", Api: standards.CONNECT} + + for name, tc := range map[string]struct { + endpoints []*basev0.Endpoint + want *basev0.Endpoint + refused string + }{ + "gRPC and Connect alone": {endpoints: []*basev0.Endpoint{grpc, connect}}, + "REST": {endpoints: []*basev0.Endpoint{rest, grpc}, want: rest}, + "plain HTTP": {endpoints: []*basev0.Endpoint{http, grpc, connect}, want: http}, + "two listeners": {endpoints: []*basev0.Endpoint{rest, http}, refused: "declares 2 HTTP endpoints (probes, rest)"}, + } { + t.Run(name, func(t *testing.T) { + got, err := findHTTPEndpoint(ctx, tc.endpoints) + if tc.refused != "" { + require.ErrorContains(t, err, tc.refused) + return + } + require.NoError(t, err) + require.Equal(t, tc.want, got) + }) + } +} + +// createConnectOnlyService scaffolds a gRPC service and rewrites its manifest +// into a Connect-only one: connect-endpoint on, and the gRPC and Connect +// endpoints in place of the scaffold's REST endpoint. +func createConnectOnlyService(t *testing.T) (*Builder, string) { + t.Helper() + builder, root := createServiceForTest(t, map[string]*agentv0.Answer{ + HotReload: confirmAnswer(false), + PublicEndpoint: confirmAnswer(false), + GRPCServer: confirmAnswer(true), + }) + manifest := filepath.Join(root, "service.codefly.yaml") + content, err := os.ReadFile(manifest) + require.NoError(t, err) + var document map[string]any + require.NoError(t, yaml.Unmarshal(content, &document)) + document["spec"].(map[string]any)["connect-endpoint"] = true + document["endpoints"] = []map[string]any{ + {"name": "grpc", "api": "grpc", "visibility": "private"}, + {"name": "connect", "api": "connect", "visibility": "private"}, + } + content, err = yaml.Marshal(document) + require.NoError(t, err) + require.NoError(t, os.WriteFile(manifest, content, 0o644)) + return builder, root +} + +func localEnvironment(t *testing.T) *basev0.Environment { + t.Helper() + environment, err := resources.LocalEnvironment().Proto() + require.NoError(t, err) + return environment +} + +func endpointNames(endpoints []*basev0.Endpoint) []string { + var names []string + for _, endpoint := range endpoints { + names = append(names, endpoint.GetName()) + } + slices.Sort(names) + return names +} + +// TestConnectOnlyServiceLoads is the builder half of the regression: Load +// refused every service without a REST endpoint ("no rest endpoint found"). +// The runtime loads it too, and Deploy, which probes the HTTP listener, says +// that it has none rather than that the health it would read is missing. +func TestConnectOnlyServiceLoads(t *testing.T) { + ctx := context.Background() + created, _ := createConnectOnlyService(t) + identity, err := created.Identity.Proto() + require.NoError(t, err) + + builder := NewBuilder(NewService()) + loaded, err := builder.Load(ctx, &builderv0.LoadRequest{Identity: identity}) + require.NoError(t, err) + require.Equal(t, builderv0.LoadStatus_READY, loaded.GetState().GetState(), loaded.GetState().GetMessage()) + require.Equal(t, []string{"connect", "grpc"}, endpointNames(loaded.GetEndpoints())) + require.Nil(t, builder.FastAPI.HTTPEndpoint) + require.True(t, builder.FastAPI.Settings.ConnectEndpoint) + + _, err = builder.Deploy(ctx, &builderv0.DeploymentRequest{}) + require.ErrorContains(t, err, "the service declares none") + + runtime := NewRuntime(NewService()) + t.Cleanup(runtime.Base.StopWatcher) + resp, err := runtime.Load(ctx, &runtimev0.LoadRequest{Identity: identity, Environment: localEnvironment(t), DisableCatch: true}) + require.NoError(t, err) + require.NotEqual(t, runtimev0.LoadStatus_ERROR, resp.GetStatus().GetState(), resp.GetStatus().GetMessage()) + require.Nil(t, runtime.FastAPI.HTTPEndpoint) + require.Equal(t, "grpc", runtime.FastAPI.GRPCEndpoint.GetName()) + require.Equal(t, "connect", runtime.FastAPI.ConnectEndpoint.GetName()) +} + +// TestConnectOnlyServiceSyncs runs the builder's Sync on a loaded Connect-only +// service: the gRPC bindings and the Connect face are generated with no REST +// endpoint in sight. The bindings come from Buf in its companion container. +func TestConnectOnlyServiceSyncs(t *testing.T) { + if testing.Short() { + t.Skip("generates the gRPC bindings with Buf in a container") + } + if err := exec.Command("docker", "info").Run(); err != nil { + t.Skipf("docker is not available, Buf cannot run: %v", err) + } + ctx := context.Background() + created, root := createConnectOnlyService(t) + identity, err := created.Identity.Proto() + require.NoError(t, err) + + builder := NewBuilder(NewService()) + loaded, err := builder.Load(ctx, &builderv0.LoadRequest{Identity: identity}) + require.NoError(t, err) + require.Equal(t, builderv0.LoadStatus_READY, loaded.GetState().GetState(), loaded.GetState().GetMessage()) + + synced, err := builder.Sync(ctx, &builderv0.SyncRequest{}) + require.NoError(t, err) + require.NotEqual(t, builderv0.SyncStatus_ERROR, synced.GetState().GetState(), synced.GetState().GetMessage()) + require.FileExists(t, filepath.Join(root, connectFacePath)) + generated, err := filepath.Glob(filepath.Join(root, "code", "src", "rpc", "_generated", "*_pb2_grpc.py")) + require.NoError(t, err) + require.NotEmpty(t, generated, "Sync wrote no gRPC bindings") +} + +// connectOnlyMappings are the proposed mappings of a Connect-only service: its +// gRPC and Connect endpoints, each on a free loopback port. +func connectOnlyMappings(t *testing.T, grpcEndpoint, connectEndpoint *basev0.Endpoint) ([]*basev0.NetworkMapping, uint16) { + t.Helper() + mapping := func(endpoint *basev0.Endpoint, address func(uint16) string) (*basev0.NetworkMapping, uint16) { + port := freePort(t) + return &basev0.NetworkMapping{Endpoint: endpoint, Instances: []*basev0.NetworkInstance{{ + Access: resources.NewNativeNetworkAccess(), Hostname: "localhost", Port: uint32(port), Address: address(port), + }}}, port + } + grpcMapping, _ := mapping(grpcEndpoint, func(port uint16) string { return fmt.Sprintf("localhost:%d", port) }) + connectMapping, connectPort := mapping(connectEndpoint, func(port uint16) string { return fmt.Sprintf("http://localhost:%d", port) }) + return []*basev0.NetworkMapping{grpcMapping, connectMapping}, connectPort +} + +// TestConnectOnlyStartPublishesNoHTTPListener: Init resolves no HTTP endpoint +// for a service that declares none, so Start binds the app's HTTP listener to a +// loopback port nobody is told about and waits on the Connect face. +func TestConnectOnlyStartPublishesNoHTTPListener(t *testing.T) { + isolateWorkingDir(t) + ctx := context.Background() + runtime, req := initTestRuntime(t) + env := &fakeRunnerEnvironment{} + runtime.runnerEnvironment = env + runtime.cacheLocation = t.TempDir() + + grpcEndpoint := &basev0.Endpoint{Module: "mod", Service: "probe", Name: "grpc", Api: standards.GRPC} + connectEndpoint := &basev0.Endpoint{Module: "mod", Service: "probe", Name: "connect", Api: standards.CONNECT} + runtime.FastAPI.HTTPEndpoint = nil + runtime.FastAPI.GRPCEndpoint = grpcEndpoint + runtime.FastAPI.ConnectEndpoint = connectEndpoint + var connectPort uint16 + req.ProposedNetworkMappings, connectPort = connectOnlyMappings(t, grpcEndpoint, connectEndpoint) + var asked []string + runtime.answers = func(_ context.Context, target string) error { + asked = append(asked, target) + return nil + } + + initialized, err := runtime.Init(ctx, req) + require.NoError(t, err) + require.Equal(t, runtimev0.InitStatus_READY, initialized.GetStatus().GetState(), initialized.GetStatus().GetMessage()) + + started, err := runtime.Start(ctx, &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_STARTED, started.GetStatus().GetState(), started.GetStatus().GetMessage()) + t.Cleanup(func() { _, _ = runtime.Stop(context.Background(), &runtimev0.StopRequest{}) }) + args := uvicornProc(t, env).args + require.Equal(t, "0", uvicornPort(t, env)) + require.Equal(t, "127.0.0.1", args[slices.Index(args, "--host")+1]) + require.Equal(t, []string{fmt.Sprintf("http://localhost:%d/", connectPort)}, asked) +} + +// TestServeTargetWithoutHTTPEndpoint: the Connect face answers HTTP, so it is +// what Start waits on; a gRPC server alone is waited on for its handshake. +func TestServeTargetWithoutHTTPEndpoint(t *testing.T) { + runtime := NewRuntime(NewService()) + runtime.grpcAddress = "localhost:9090" + runtime.connectAddress = "http://localhost:8081" + name, target, _ := runtime.serveTarget() + require.Equal(t, "Connect face", name) + require.Equal(t, "http://localhost:8081/", target) + + runtime.connectAddress = "" + name, target, _ = runtime.serveTarget() + require.Equal(t, "gRPC server", name) + require.Equal(t, "localhost:9090", target) + + runtime.grpcAddress = "" + _, target, _ = runtime.serveTarget() + require.Empty(t, target, "nothing to wait on") +} + +// TestGRPCAnswersNeedsAHandshake: a gRPC server answers; a listener that +// accepts and hangs up, as an engine's proxy does before anything listens +// behind it, does not. +func TestGRPCAnswersNeedsAHandshake(t *testing.T) { + ctx := context.Background() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + server := grpc.NewServer() + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + require.NoError(t, grpcAnswers(ctx, listener.Addr().String())) + require.NoError(t, grpcAnswers(ctx, "http://"+listener.Addr().String()), "a scheme in the address is not part of the target") + + hangup, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = hangup.Close() }) + go func() { + for { + conn, err := hangup.Accept() + if err != nil { + return + } + _ = conn.Close() + } + }() + require.Error(t, grpcAnswers(ctx, hangup.Addr().String())) +} + +// generateScaffoldStubs writes the gRPC bindings Sync generates with Buf the +// way the scaffold's own tests/rpc/conftest.py does when run directly: same +// plugins, no network plugin registry. +func generateScaffoldStubs(t *testing.T, code string) { + t.Helper() + require.NoError(t, os.MkdirAll(filepath.Join(code, "src", "rpc", "_generated"), 0o755)) + protoc := exec.Command("uv", "run", "python", "-m", "grpc_tools.protoc", + "-I../proto", "--python_out=src/rpc/_generated", "--grpc_python_out=src/rpc/_generated", "../proto/api.proto") + protoc.Dir = code + out, err := protoc.CombinedOutput() + require.NoError(t, err, "%s", out) +} + +// TestConnectOnlyServiceStarts drives the runtime through Load, Init and Start +// on a scaffolded Connect-only service with a real uv: STARTED arrives, and +// the face it waited on answers a Connect call. +func TestConnectOnlyServiceStarts(t *testing.T) { + if testing.Short() { + t.Skip("boots a scaffolded service with uv") + } + if _, err := exec.LookPath("uv"); err != nil { + t.Skip("uv is not installed: the scaffolded service cannot run") + } + ctx := context.Background() + created, root := createConnectOnlyService(t) + code := filepath.Join(root, "code") + identity, err := created.Identity.Proto() + require.NoError(t, err) + // The face Sync writes, from the loaded manifest's settings. + builder := NewBuilder(NewService()) + _, err = builder.Load(ctx, &builderv0.LoadRequest{Identity: identity}) + require.NoError(t, err) + require.NoError(t, builder.syncConnectFace()) + generateScaffoldStubs(t, code) + + runtime := NewRuntime(NewService()) + t.Cleanup(func() { _, _ = runtime.Destroy(context.Background(), &runtimev0.DestroyRequest{}) }) + loaded, err := runtime.Load(ctx, &runtimev0.LoadRequest{Identity: identity, Environment: localEnvironment(t), DisableCatch: true}) + require.NoError(t, err) + require.NotEqual(t, runtimev0.LoadStatus_ERROR, loaded.GetStatus().GetState(), loaded.GetStatus().GetMessage()) + + mappings, connectPort := connectOnlyMappings(t, runtime.FastAPI.GRPCEndpoint, runtime.FastAPI.ConnectEndpoint) + initialized, err := runtime.Init(ctx, &runtimev0.InitRequest{ + RuntimeContext: resources.NewRuntimeContextNative(), ProposedNetworkMappings: mappings, + }) + require.NoError(t, err) + require.Equal(t, runtimev0.InitStatus_READY, initialized.GetStatus().GetState(), initialized.GetStatus().GetMessage()) + + started, err := runtime.Start(ctx, &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_STARTED, started.GetStatus().GetState(), started.GetStatus().GetMessage()) + + request, err := http.NewRequest(http.MethodPost, + fmt.Sprintf("http://127.0.0.1:%d/api.EchoService/Echo", connectPort), + strings.NewReader(`{"message":"connect only"}`)) + require.NoError(t, err) + request.Header.Set("Content-Type", "application/json") + client := http.Client{Timeout: 10 * time.Second} + response, err := client.Do(request) + require.NoError(t, err) + body, err := io.ReadAll(response.Body) + _ = response.Body.Close() + require.NoError(t, err) + require.Equal(t, http.StatusOK, response.StatusCode, "%s", body) + require.JSONEq(t, `{"message":"connect only"}`, string(body)) + + _, err = runtime.Stop(ctx, &runtimev0.StopRequest{}) + require.NoError(t, err) +} diff --git a/connect_test.go b/connect_test.go index 3eb8e0c..da96633 100644 --- a/connect_test.go +++ b/connect_test.go @@ -185,16 +185,7 @@ func TestGeneratedMainServesConnect(t *testing.T) { GRPCServer: confirmAnswer(true), }) code := filepath.Join(root, "code") - // The stubs Sync generates with Buf, generated the way the scaffold's own - // tests/rpc/conftest.py does when run directly: same plugins, no network - // plugin registry. - generated := filepath.Join(code, "src", "rpc", "_generated") - require.NoError(t, os.MkdirAll(generated, 0o755)) - protoc := exec.Command("uv", "run", "python", "-m", "grpc_tools.protoc", - "-I../proto", "--python_out=src/rpc/_generated", "--grpc_python_out=src/rpc/_generated", "../proto/api.proto") - protoc.Dir = code - out, err := protoc.CombinedOutput() - require.NoError(t, err, "%s", out) + generateScaffoldStubs(t, code) builder.FastAPI.Settings.ConnectEndpoint = true require.NoError(t, builder.syncConnectFace()) diff --git a/init_test.go b/init_test.go index 060b1d8..e24474a 100644 --- a/init_test.go +++ b/init_test.go @@ -57,7 +57,7 @@ func initTestRuntime(t *testing.T) (*Runtime, *runtimev0.InitRequest) { runtime.Service.SourceLocation = source endpoint := &basev0.Endpoint{Module: "mod", Service: "probe", Name: "rest", Api: standards.REST} - runtime.FastAPI.RestEndpoint = endpoint + runtime.FastAPI.HTTPEndpoint = endpoint t.Cleanup(runtime.Base.StopWatcher) return runtime, &runtimev0.InitRequest{ diff --git a/main.go b/main.go index 0464638..81b6712 100644 --- a/main.go +++ b/main.go @@ -9,6 +9,8 @@ import ( "context" "embed" "fmt" + "sort" + "strings" "github.com/codefly-dev/core/agents" "github.com/codefly-dev/core/agents/contract" @@ -56,8 +58,8 @@ const ( const defaultProtoPath = standards.ProtoPath // GRPCServerSettings configures the optional service-owned grpc.aio listener. -// Disabled by default: an unset grpc-server block leaves the service REST-only -// and its generated layout unchanged. +// Disabled by default: an unset grpc-server block leaves the service with no +// gRPC listener and its generated layout unchanged. type GRPCServerSettings struct { Enabled bool `yaml:"enabled"` @@ -155,7 +157,7 @@ const uvImage = "ghcr.io/astral-sh/uv:" + uvVersion // *services.Base: Wool, Logger, Location, Identity, …) are promoted. // // FastAPI-specific: a richer Settings (with HotReload / PublicEndpoint) -// and the REST endpoint handle. FastAPI source lives under ./code — the +// and the endpoint handles. FastAPI source lives under ./code — the // Builder/Runtime set Service.SourceLocation to s.Local("code") during // Load. type Service struct { @@ -166,10 +168,14 @@ type Service struct { // the FastAPI settings; `s.Service.Settings` is the generic view. Settings *Settings - RestEndpoint *v0.Endpoint + // HTTPEndpoint is the endpoint the app's HTTP listener (uvicorn) serves: + // the service's REST endpoint, whose routes src/openapi.py documents, or a + // plain HTTP one, which carries no API document (probes, a solution's own + // routes). Nil for a service whose faces are gRPC and Connect alone. + HTTPEndpoint *v0.Endpoint // GRPCEndpoint is the service-owned gRPC endpoint, present only when - // Settings.GRPCServer.Enabled. Nil keeps the REST-only path untouched. + // Settings.GRPCServer.Enabled. GRPCEndpoint *v0.Endpoint // ConnectEndpoint is the Connect-family endpoint, present only when @@ -177,6 +183,27 @@ type Service struct { ConnectEndpoint *v0.Endpoint } +// findHTTPEndpoint returns the one endpoint the app's HTTP listener serves: a +// REST or a plain HTTP endpoint. None is not an error — a service whose faces +// are gRPC and Connect alone declares no HTTP endpoint — but two are, because +// the process has one HTTP listener and could serve only one of them. +func findHTTPEndpoint(ctx context.Context, endpoints []*v0.Endpoint) (*v0.Endpoint, error) { + found := resources.FindEndpointsByAPI(ctx, standards.REST, endpoints) + found = append(found, resources.FindEndpointsByAPI(ctx, standards.HTTP, endpoints)...) + switch len(found) { + case 0: + return nil, nil + case 1: + return found[0], nil + } + names := make([]string, 0, len(found)) + for _, endpoint := range found { + names = append(names, endpoint.GetName()) + } + sort.Strings(names) + return nil, fmt.Errorf("the app serves one HTTP listener, and the service declares %d HTTP endpoints (%s): keep one with api: rest or api: http", len(found), strings.Join(names, ", ")) +} + // connectSettingError refuses a Connect face with no gRPC server to transcode // onto: the face serves the gRPC server's API, never one of its own. func (s *Settings) connectSettingError() error { @@ -206,9 +233,10 @@ func (s *Service) GetAgentInformation(ctx context.Context, _ *agentv0.AgentInfor Toolchains: []agentv0.Toolchain_Type{agentv0.Toolchain_PYTHON}, HotReload: true, Languages: []agentv0.Language_Type{agentv0.Language_PYTHON}, - // The agent can serve HTTP always and gRPC when a service opts in via - // grpc-server (and Connect beside it, over HTTP); advertising both - // declares the capability, not that every service exposes both. + // The agent serves HTTP when a service declares a REST or HTTP endpoint + // and gRPC when it opts in via grpc-server (and Connect beside it, over + // HTTP); advertising both declares the capability, not that every + // service exposes both. Protocols: []agentv0.Protocol_Type{agentv0.Protocol_HTTP, agentv0.Protocol_GRPC}, ReadMe: readme, }.Build() diff --git a/main_test.go b/main_test.go index ebd4180..5644476 100644 --- a/main_test.go +++ b/main_test.go @@ -138,7 +138,7 @@ func testRun(t *testing.T, runtime *Runtime, ctx context.Context, identity *base // confusing empty-network-mappings error further down. require.Equal(t, runtimev0.InitStatus_READY, init.GetStatus().GetState(), init.GetStatus().GetMessage()) - instance, err := resources.FindNetworkInstanceInNetworkMappings(ctx, init.NetworkMappings, runtime.FastAPI.RestEndpoint, resources.NewNativeNetworkAccess()) + instance, err := resources.FindNetworkInstanceInNetworkMappings(ctx, init.NetworkMappings, runtime.FastAPI.HTTPEndpoint, resources.NewNativeNetworkAccess()) require.NoError(t, err) _, err = runtime.Start(ctx, &runtimev0.StartRequest{}) diff --git a/probes_test.go b/probes_test.go index 4bde5e1..9e060af 100644 --- a/probes_test.go +++ b/probes_test.go @@ -183,7 +183,7 @@ func TestBuilderLoadsAuthoredTLSAndHealthBeforeProjection(t *testing.T) { loaded := NewBuilder(NewService()) _, err = loaded.Load(t.Context(), &builderv0.LoadRequest{Identity: identity}) require.NoError(t, err) - endpoint := loaded.FastAPI.RestEndpoint + endpoint := loaded.FastAPI.HTTPEndpoint require.True(t, resources.EndpointSecured(endpoint)) require.Equal(t, "/readyz", endpoint.GetHealth().GetReadiness().GetHttp().GetPath()) probes, err := deploymentProbes(endpoint, false) diff --git a/runtime.go b/runtime.go index ddc7078..6f3982c 100644 --- a/runtime.go +++ b/runtime.go @@ -28,6 +28,10 @@ import ( "github.com/codefly-dev/core/wool" pythonruntime "github.com/codefly-dev/service-python/pkg/runtime" + + "google.golang.org/grpc" + "google.golang.org/grpc/connectivity" + "google.golang.org/grpc/credentials/insecure" ) // Runtime is the FastAPI specialization of the generic Python Runtime. @@ -37,7 +41,7 @@ import ( // *pythonruntime.Runtime — inherits Test (uv run pytest), Lint (uv run ruff), // Build (no-op), Information, and the services.Base // chain via *pythonservice.Service promotion. -// FastAPI — fastapi-specific state (RestEndpoint, HotReload +// FastAPI — fastapi-specific state (endpoint handles, HotReload // setting) accessed explicitly as s.FastAPI.X. // // Overridden methods: Load, Init, Start, Stop, Destroy — fastapi adds Docker @@ -132,11 +136,15 @@ type Runtime struct { // filtered back out of the process environment. appliedOverrides map[string]int + // port is the mapped port of the HTTP endpoint, which uvicorn binds. Zero + // when the service declares none: uvicorn then binds a loopback port the + // system picks, which no endpoint publishes (uvicornArgs). port uint16 // serveAddress is the address Init resolved for the HTTP endpoint, as the // host reaches it. Start answers STARTED only once an HTTP request to it gets - // a response. Empty (no Init) skips the wait. + // a response. Empty when the service declares no HTTP endpoint, and then + // Start waits on the Connect face, else on the gRPC server (awaitServing). serveAddress string // serveTimeout bounds that wait; zero means defaultServeTimeout. @@ -149,13 +157,16 @@ type Runtime struct { // grpcPort is the mapped port for the service-owned gRPC listener, passed // to the Python process as CODEFLY_GRPC_PORT. Zero when gRPC is disabled. grpcPort uint16 + // grpcAddress is the address Init resolved for the gRPC server, which Start + // waits on when the service declares neither an HTTP endpoint nor a Connect + // face. Empty when gRPC is disabled. + grpcAddress string // connectPort is the mapped port for the Connect face, passed to the // Python process as CODEFLY_CONNECT_PORT. Zero when Connect is disabled. connectPort uint16 // connectAddress is the address Init resolved for the Connect face, which - // Start requires to answer once the HTTP endpoint does. Empty when Connect - // is disabled. + // Start requires to answer. Empty when Connect is disabled. connectAddress string cacheLocation string @@ -198,7 +209,7 @@ func (s *Runtime) Load(ctx context.Context, req *runtimev0.LoadRequest) (*runtim } s.Endpoints = endpoints - s.FastAPI.RestEndpoint, err = resources.FindRestEndpoint(ctx, s.Endpoints) + s.FastAPI.HTTPEndpoint, err = findHTTPEndpoint(ctx, s.Endpoints) if err != nil { return s.Base.Runtime.LoadError(err) } @@ -296,11 +307,13 @@ func (s *Runtime) CreateRunnerEnvironment(ctx context.Context) (runners.RunnerEn // host where the user isn't root (e.g. Linux CI). dockerEnv.WithUser(fmt.Sprintf("%d:%d", os.Getuid(), os.Getgid())) - instance, err := resources.FindNetworkInstanceInNetworkMappings(ctx, s.NetworkMappings, s.FastAPI.RestEndpoint, resources.NewNativeNetworkAccess()) - if err != nil { - return nil, "", s.Wool.Wrapf(err, "cannot find network instance") + if s.FastAPI.HTTPEndpoint != nil { + instance, err := resources.FindNetworkInstanceInNetworkMappings(ctx, s.NetworkMappings, s.FastAPI.HTTPEndpoint, resources.NewNativeNetworkAccess()) + if err != nil { + return nil, "", s.Wool.Wrapf(err, "cannot find network instance") + } + dockerEnv.WithPort(ctx, uint16(instance.Port)) } - dockerEnv.WithPort(ctx, uint16(instance.Port)) if s.FastAPI.GRPCEndpoint != nil { grpcInstance, grpcErr := resources.FindNetworkInstanceInNetworkMappings(ctx, s.NetworkMappings, s.FastAPI.GRPCEndpoint, resources.NewNativeNetworkAccess()) @@ -486,21 +499,24 @@ func (s *Runtime) Init(ctx context.Context, req *runtimev0.InitRequest) (*runtim return s.Base.Runtime.InitError(err) } - net, err := resources.FindNetworkInstanceInNetworkMappings(ctx, s.NetworkMappings, s.FastAPI.RestEndpoint, resources.NewNativeNetworkAccess()) - if err != nil { - return s.Base.Runtime.InitError(err) + s.port, s.serveAddress = 0, "" + if s.FastAPI.HTTPEndpoint != nil { + net, err := resources.FindNetworkInstanceInNetworkMappings(ctx, s.NetworkMappings, s.FastAPI.HTTPEndpoint, resources.NewNativeNetworkAccess()) + if err != nil { + return s.Base.Runtime.InitError(err) + } + s.Infof("will run on %s", net.Address) + s.port = uint16(net.Port) + s.serveAddress = net.Address } - s.Infof("will run on %s", net.Address) - s.port = uint16(net.Port) - s.serveAddress = net.Address - if s.FastAPI.GRPCEndpoint != nil { grpcNet, grpcErr := resources.FindNetworkInstanceInNetworkMappings(ctx, s.NetworkMappings, s.FastAPI.GRPCEndpoint, resources.NewNativeNetworkAccess()) if grpcErr != nil { return s.Base.Runtime.InitError(grpcErr) } s.grpcPort = uint16(grpcNet.Port) + s.grpcAddress = grpcNet.Address s.Infof("grpc will run on %s", grpcNet.Address) } @@ -729,33 +745,51 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt return resp, err } -// defaultServeTimeout bounds how long Start waits for the HTTP endpoint to -// answer. It covers `uv run` resolving the environment, the app import and its +// defaultServeTimeout bounds how long Start waits for the service to answer. +// It covers `uv run` resolving the environment, the app import and its // lifespan startup; a service that has not answered by then is not coming up. const defaultServeTimeout = 3 * time.Minute // servePollInterval is the pause between two attempts to reach the endpoint. const servePollInterval = 250 * time.Millisecond -// awaitServing returns once an HTTP request to the service's endpoint gets a -// response — any status: readiness here means the endpoint accepts and serves -// connections, and what the app answers on "/" is its own business. A bare TCP -// accept is not enough: in a container runtime the published port is accepted -// by the engine's proxy before anything listens behind it. It fails when the -// process exits first, when ctx ends, or after the serve timeout. +// serveTarget is the listener Start waits on before it answers STARTED, and +// how one attempt reaches it: the HTTP endpoint when the service declares one, +// else the Connect face, else the gRPC server. Empty when there is none (no +// Init), which skips the wait. +func (s *Runtime) serveTarget() (name, target string, answers func(context.Context, string) error) { + switch { + case s.serveAddress != "": + name, target, answers = "HTTP endpoint", httpTarget(s.serveAddress), httpAnswers + case s.connectAddress != "": + name, target, answers = "Connect face", httpTarget(s.connectAddress), httpAnswers + case s.grpcAddress != "": + name, target, answers = "gRPC server", s.grpcAddress, grpcAnswers + default: + return "", "", nil + } + if s.answers != nil { + answers = s.answers + } + return name, target, answers +} + +// awaitServing returns once the service's listener answers (serveTarget). An +// HTTP listener answers with any status: readiness here means the endpoint +// accepts and serves connections, and what the app answers on "/" is its own +// business. A bare TCP accept is not enough: in a container runtime the +// published port is accepted by the engine's proxy before anything listens +// behind it. It fails when the process exits first, when ctx ends, or after the +// serve timeout. func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error { - if s.serveAddress == "" { + name, target, answers := s.serveTarget() + if target == "" { return nil } - target := httpTarget(s.serveAddress) timeout := s.serveTimeout if timeout <= 0 { timeout = defaultServeTimeout } - answers := s.answers - if answers == nil { - answers = httpAnswers - } deadline := time.NewTimer(timeout) defer deadline.Stop() started := time.Now() @@ -763,18 +797,21 @@ func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error for { err := answers(ctx, target) if err == nil { - s.Infof("fastapi app serving on %s after %s", s.serveAddress, time.Since(started).Round(100*time.Millisecond)) + s.Infof("fastapi app serving its %s at %s after %s", name, target, time.Since(started).Round(100*time.Millisecond)) + if s.serveAddress == "" { + return nil + } return s.awaitConnect(ctx, answers) } last = err if !s.runnerAlive(ctx, handle) { - return fmt.Errorf("fastapi app exited before its HTTP endpoint at %s answered", s.serveAddress) + return fmt.Errorf("fastapi app exited before its %s at %s answered", name, target) } select { case <-ctx.Done(): - return fmt.Errorf("start abandoned before the HTTP endpoint at %s answered: %w", s.serveAddress, ctx.Err()) + return fmt.Errorf("start abandoned before the %s at %s answered: %w", name, target, ctx.Err()) case <-deadline.C: - return fmt.Errorf("the HTTP endpoint at %s did not answer within %s: %v", s.serveAddress, timeout, last) + return fmt.Errorf("the %s at %s did not answer within %s: %v", name, target, timeout, last) case <-time.After(servePollInterval): } } @@ -842,11 +879,47 @@ func httpAnswers(ctx context.Context, target string) error { return response.Body.Close() } +// grpcAnswers reports whether a gRPC server at address completes the HTTP/2 +// handshake within one bounded attempt. The connection is ready only once the +// server's own SETTINGS frame arrives, which an engine's proxy accepting the +// port on the server's behalf cannot send. +func grpcAnswers(ctx context.Context, address string) error { + if _, hostPort, found := strings.Cut(address, "://"); found { + address = hostPort + } + conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return err + } + defer func() { _ = conn.Close() }() + ctx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() + conn.Connect() + for { + state := conn.GetState() + if state == connectivity.Ready { + return nil + } + if !conn.WaitForStateChange(ctx, state) { + return fmt.Errorf("no gRPC handshake at %s (%s)", address, state) + } + } +} + // uvicornArgs renders the `uv run` invocation. --reload is conditional: it // makes uvicorn fork a supervisor that re-executes the app on source edits, // which is not what a service configured without hot reload asked for. +// +// Port zero is a service with no HTTP endpoint. Its app still runs under +// uvicorn, because the app's lifespan is what starts its gRPC server and +// Connect face, but its HTTP listener is no endpoint: it binds a loopback port +// the system picks and no one is told. func uvicornArgs(port uint16, hotReload bool) []string { - args := []string{"run", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", fmt.Sprintf("%d", port), "--log-config", uvicornLogConfigFile} + host := "0.0.0.0" + if port == 0 { + host = "127.0.0.1" + } + args := []string{"run", "uvicorn", "src.main:app", "--host", host, "--port", fmt.Sprintf("%d", port), "--log-config", uvicornLogConfigFile} if hotReload { args = append(args, "--reload") } @@ -1273,9 +1346,6 @@ func (s *Runtime) destroyContainerByIdentity(ctx context.Context) error { } func (s *Runtime) EventHandler(event code.Change) error { - if strings.Contains(event.Path, "api.json") { - return nil - } if strings.HasSuffix(event.Path, ".py") { // uvicorn --reload picks these up; no action needed here. return nil diff --git a/templates/agent/README.md.tmpl b/templates/agent/README.md.tmpl index e82d45b..9aa9b72 100644 --- a/templates/agent/README.md.tmpl +++ b/templates/agent/README.md.tmpl @@ -16,7 +16,7 @@ Some of the things this agent provides for you: ## Declared container health -Kubernetes probes follow the authored REST endpoint's health predicates and transport. HTTP checks preserve accepted statuses and bounded body containment; redirects are not followed. Secured container-local checks authenticate the listener against the exact certificate mounted at `UVICORN_SSL_CERTFILE` before sending HTTP data. The isolated Python probe imports no application code and emits no response content or certificate material. +Kubernetes probes follow the health predicates of the authored HTTP endpoint (`api: rest` or `api: http`) and transport. HTTP checks preserve accepted statuses and bounded body containment; redirects are not followed. Secured container-local checks authenticate the listener against the exact certificate mounted at `UVICORN_SSL_CERTFILE` before sending HTTP data. The isolated Python probe imports no application code and emits no response content or certificate material. Startup, readiness and liveness are separate policies. The current Core Kubernetes manifest profile requires all three, so deployment refuses absent startup/liveness declarations instead of copying readiness into a restart policy. New scaffolds explicitly declare `/version` readiness and transport startup/liveness. Existing services must author their own policies. Whole-second timing is required by Kubernetes; predicates or additional listener health that this agent cannot preserve are refused rather than silently omitted.