diff --git a/Dockerfile b/Dockerfile index d73707b..83ef8b1 100644 --- a/Dockerfile +++ b/Dockerfile @@ -91,6 +91,12 @@ RUN go build -trimpath -ldflags="-s -w \ -X main.BuildDate=${BUILD_DATE}" \ -o /out/go-order-actor ./cmd/order +RUN go build -trimpath -ldflags="-s -w \ + -X main.Version=${VERSION} \ + -X main.GitCommit=${GIT_COMMIT} \ + -X main.BuildDate=${BUILD_DATE}" \ + -o /out/go-profile-actor ./cmd/profile + ############################ # Runtime Stage ############################ @@ -103,6 +109,7 @@ COPY --from=build /out/go-checkout-actor /go-checkout-actor COPY --from=build /out/go-cart-backoffice /go-cart-backoffice COPY --from=build /out/go-cart-inventory /go-cart-inventory COPY --from=build /out/go-order-actor /go-order-actor +COPY --from=build /out/go-profile-actor /go-profile-actor # Document (not expose forcibly) typical ports: 8080 (HTTP), 1337 (gRPC) EXPOSE 8080 1337 diff --git a/cmd/cart/main.go b/cmd/cart/main.go index 38cb182..da3dc47 100644 --- a/cmd/cart/main.go +++ b/cmd/cart/main.go @@ -18,7 +18,6 @@ import ( "git.k6n.net/mats/go-cart-actor/pkg/promotions" promotionmcp "git.k6n.net/mats/go-cart-actor/pkg/promotions/mcp" "git.k6n.net/mats/go-cart-actor/internal/ucp" - "git.k6n.net/mats/go-cart-actor/pkg/profile" "git.k6n.net/mats/go-cart-actor/pkg/proxy" "git.k6n.net/mats/go-cart-actor/pkg/voucher" "git.k6n.net/mats/go-redis-inventory/pkg/inventory" @@ -297,38 +296,6 @@ func main() { syncedServer.Serve(mux) - // ── Profile / Customer UCP endpoint ─────────────────────────────────── - profileReg := profile.NewProfileMutationRegistry() - profileDir := getEnv("PROFILE_DIR", "data/profiles") - _ = os.MkdirAll(profileDir, 0755) - profileStorage := actor.NewDiskStorage[profile.ProfileGrain](profileDir, profileReg) - - profilePoolConfig := actor.GrainPoolConfig[profile.ProfileGrain]{ - MutationRegistry: profileReg, - Storage: profileStorage, - Spawn: func(ctx context.Context, id uint64) (actor.Grain[profile.ProfileGrain], error) { - ret := profile.NewProfileGrain(id, time.Now()) - err := profileStorage.LoadEvents(ctx, id, ret) - return ret, err - }, - TTL: 10 * time.Minute, - PoolSize: 65535, - Hostname: podIp, - } - - profilePool, poolErr := actor.NewSimpleGrainPool(profilePoolConfig) - if poolErr != nil { - log.Fatalf("Error creating profile pool: %v\n", poolErr) - } - - // UCP Customer REST adapter - customerUCP := ucp.CustomerHandler(profilePool) - if signer := loadUCPCartSigner(); signer != nil { - customerUCP = ucp.WithSigning(customerUCP, signer) - log.Print("ucp customer signing enabled") - } - mux.Handle("/ucp/v1/customers/", http.StripPrefix("/ucp/v1/customers", customerUCP)) - // Promotion MCP edge: list/get/upsert/status/delete promotions and preview // discounts. Mutations persist to data/promotions.json and are picked up by // the cart's totals processor on the next mutation (it reads a fresh snapshot). diff --git a/cmd/profile/main.go b/cmd/profile/main.go new file mode 100644 index 0000000..1ee542b --- /dev/null +++ b/cmd/profile/main.go @@ -0,0 +1,156 @@ +package main + +import ( + "context" + "fmt" + "log" + "net" + "net/http" + "os" + "os/signal" + "time" + + "git.k6n.net/mats/go-cart-actor/internal/ucp" + "git.k6n.net/mats/go-cart-actor/pkg/actor" + "git.k6n.net/mats/go-cart-actor/pkg/profile" + "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" +) + +var podIp = os.Getenv("POD_IP") +var name = os.Getenv("POD_NAME") + +// loadUCPProfileSigner loads the UCP ECDSA signing key from the path specified in +// the UCP_SIGNING_KEY_PATH environment variable. Returns nil if unset or unreadable. +func loadUCPProfileSigner() *ucp.SigningConfig { + path := os.Getenv("UCP_SIGNING_KEY_PATH") + if path == "" { + return nil + } + pemData, err := os.ReadFile(path) + if err != nil { + log.Printf("ucp signing: cannot read key %s: %v", path, err) + return nil + } + cfg, err := ucp.NewSigningConfigFromPEM(pemData, "k6n-ecdsa-2026") + if err != nil { + log.Printf("ucp signing: invalid key at %s: %v", path, err) + return nil + } + return cfg +} + +// getEnv returns the value of the environment variable named by key, or def if unset/empty. +func getEnv(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +func main() { + reg := profile.NewProfileMutationRegistry() + + profileDir := getEnv("PROFILE_DIR", "data/profiles") + _ = os.MkdirAll(profileDir, 0755) + diskStorage := actor.NewDiskStorage[profile.ProfileGrain](profileDir, reg) + + poolConfig := actor.GrainPoolConfig[profile.ProfileGrain]{ + MutationRegistry: reg, + Storage: diskStorage, + Spawn: func(ctx context.Context, id uint64) (actor.Grain[profile.ProfileGrain], error) { + ret := profile.NewProfileGrain(id, time.Now()) + err := diskStorage.LoadEvents(ctx, id, ret) + return ret, err + }, + TTL: 10 * time.Minute, + PoolSize: 65535, + Hostname: podIp, + } + + pool, err := actor.NewSimpleGrainPool(poolConfig) + if err != nil { + log.Fatalf("Error creating profile pool: %v\n", err) + } + + mux := http.NewServeMux() + debugMux := http.NewServeMux() + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + + otelShutdown, err := setupOTelSDK(ctx) + if err != nil { + log.Fatalf("Unable to start otel %v", err) + } + + // UCP Customer REST adapter + customerUCP := ucp.CustomerHandler(pool) + if signer := loadUCPProfileSigner(); signer != nil { + customerUCP = ucp.WithSigning(customerUCP, signer) + log.Print("ucp customer signing enabled") + } + // StripPrefix is required because the sub-mux (CustomerHandler) registers + // routes like "GET /{id}" which expect a relative path; Go's ServeMux + // passes the full request path without stripping the prefix. + mux.Handle("/ucp/v1/customers/", http.StripPrefix("/ucp/v1/customers", customerUCP)) + + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + grainCount, capacity := pool.LocalUsage() + if grainCount >= capacity { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte("grain pool at capacity")) + return + } + if !pool.IsHealthy() { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte("pool not healthy")) + return + } + w.WriteHeader(http.StatusOK) + w.Write([]byte("ok")) + }) + mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte("ok")) + }) + mux.HandleFunc("/livez", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte("ok")) + }) + + mux.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte("1.0.0")) + }) + + srv := &http.Server{ + Addr: ":8080", + BaseContext: func(net.Listener) context.Context { return ctx }, + ReadTimeout: 10 * time.Second, + WriteTimeout: 20 * time.Second, + Handler: otelhttp.NewHandler(mux, "/"), + } + + defer func() { + fmt.Println("Shutting down profile service") + otelShutdown(context.Background()) + diskStorage.Close() + pool.Close() + }() + + srvErr := make(chan error, 1) + go func() { + srvErr <- srv.ListenAndServe() + }() + + log.Print("Profile server started at port 8080") + + go http.ListenAndServe(":8081", debugMux) + + select { + case err = <-srvErr: + log.Fatalf("Unable to start server: %v", err) + case <-ctx.Done(): + stop() + } +} diff --git a/cmd/profile/otel.go b/cmd/profile/otel.go new file mode 100644 index 0000000..fc16dc1 --- /dev/null +++ b/cmd/profile/otel.go @@ -0,0 +1,108 @@ +package main + +import ( + "context" + "errors" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc" + "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" + "go.opentelemetry.io/otel/log/global" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/sdk/log" + "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/trace" +) + +// setupOTelSDK bootstraps the OpenTelemetry pipeline. +// If it does not return an error, make sure to call shutdown for proper cleanup. +func setupOTelSDK(ctx context.Context) (func(context.Context) error, error) { + var shutdownFuncs []func(context.Context) error + var err error + + shutdown := func(ctx context.Context) error { + var err error + for _, fn := range shutdownFuncs { + err = errors.Join(err, fn(ctx)) + } + shutdownFuncs = nil + return err + } + + handleErr := func(inErr error) { + err = errors.Join(inErr, shutdown(ctx)) + } + + prop := newPropagator() + otel.SetTextMapPropagator(prop) + + tracerProvider, err := newTracerProvider() + if err != nil { + handleErr(err) + return shutdown, err + } + shutdownFuncs = append(shutdownFuncs, tracerProvider.Shutdown) + otel.SetTracerProvider(tracerProvider) + + meterProvider, err := newMeterProvider() + if err != nil { + handleErr(err) + return shutdown, err + } + shutdownFuncs = append(shutdownFuncs, meterProvider.Shutdown) + otel.SetMeterProvider(meterProvider) + + loggerProvider, err := newLoggerProvider() + if err != nil { + handleErr(err) + return shutdown, err + } + shutdownFuncs = append(shutdownFuncs, loggerProvider.Shutdown) + global.SetLoggerProvider(loggerProvider) + + return shutdown, err +} + +func newPropagator() propagation.TextMapPropagator { + return propagation.NewCompositeTextMapPropagator( + propagation.TraceContext{}, + propagation.Baggage{}, + ) +} + +func newTracerProvider() (*trace.TracerProvider, error) { + traceExporter, err := otlptracegrpc.New(context.Background()) + if err != nil { + return nil, err + } + + tracerProvider := trace.NewTracerProvider( + trace.WithBatcher(traceExporter, + trace.WithBatchTimeout(time.Second)), + ) + return tracerProvider, nil +} + +func newMeterProvider() (*metric.MeterProvider, error) { + exporter, err := otlpmetricgrpc.New(context.Background()) + if err != nil { + return nil, err + } + + provider := metric.NewMeterProvider(metric.WithReader(metric.NewPeriodicReader(exporter))) + return provider, nil +} + +func newLoggerProvider() (*log.LoggerProvider, error) { + logExporter, err := otlploggrpc.New(context.Background()) + if err != nil { + return nil, err + } + + loggerProvider := log.NewLoggerProvider( + log.WithProcessor(log.NewBatchProcessor(logExporter)), + ) + return loggerProvider, nil +}