From 679a7f6bd9732f03b1bbfc704c5f2ffc6dff30d3 Mon Sep 17 00:00:00 2001 From: Dmitrii Creed Date: Wed, 17 Jun 2026 12:11:58 +0400 Subject: [PATCH 1/2] fix(security): retry GitHub Actions OIDC token request under load The keyless-signing OIDC token was fetched with a single HTTP GET and a hard 10s client timeout. When several stacks that share one image deploy concurrently from the same push, each process races for a token and the GitHub Actions OIDC endpoint intermittently times out, surfacing "requesting OIDC token: context deadline exceeded" and failing the deploy. Retry the request with exponential backoff and full jitter (4 attempts, 1s->8s) to decorrelate concurrent callers, replace the fixed client timeout with a per-attempt context deadline (20s), and abort the loop immediately when the parent context is cancelled (returning ctx.Err() unwrapped). Only retry transient failures (transport/timeout, 408, 429, 5xx); fail fast on 4xx and on a malformed or empty 200 body. Attempts and per-attempt timeout are overridable via SC_OIDC_TOKEN_REQUEST_ATTEMPTS / _TIMEOUT. Sharing one token across the racing processes would be strictly better but is a workflow-level change outside this package; this is the in-repo defense. Signed-off-by: Dmitrii Creed --- pkg/security/context.go | 162 ++++++++++++++++---- pkg/security/context_oidc_retry_test.go | 190 ++++++++++++++++++++++++ 2 files changed, 327 insertions(+), 25 deletions(-) create mode 100644 pkg/security/context_oidc_retry_test.go diff --git a/pkg/security/context.go b/pkg/security/context.go index 16a9d35e..60fc92a8 100644 --- a/pkg/security/context.go +++ b/pkg/security/context.go @@ -5,8 +5,11 @@ import ( "encoding/json" "fmt" "io" + "math/rand" "net/http" + "net/url" "os" + "strconv" "time" ) @@ -89,43 +92,152 @@ func (e *ExecutionContext) GetOIDCToken(ctx context.Context) error { e.OIDCTokenURL = requestURL e.RequestToken = requestToken - req, err := http.NewRequestWithContext(ctx, "GET", requestURL+"&audience=sigstore", nil) + token, err := requestOIDCTokenWithRetry(ctx, requestURL, requestToken, defaultOIDCRetryPolicy(), sleepContext) if err != nil { - return fmt.Errorf("creating token request: %w", err) + return err } - req.Header.Set("Authorization", "Bearer "+requestToken) + e.OIDCToken = token + return nil + } - client := &http.Client{Timeout: 10 * time.Second} - resp, err := client.Do(req) - if err != nil { - return fmt.Errorf("requesting OIDC token: %w", err) - } - defer resp.Body.Close() + return fmt.Errorf("OIDC token not available") +} - if resp.StatusCode != http.StatusOK { - body, _ := io.ReadAll(resp.Body) - return fmt.Errorf("OIDC token request failed with status %d: %s", resp.StatusCode, string(body)) - } +type oidcRetryPolicy struct { + Attempts int + PerAttemptTimeout time.Duration + BaseBackoff time.Duration + MaxBackoff time.Duration +} - body, err := io.ReadAll(resp.Body) - if err != nil { - return fmt.Errorf("reading token response: %w", err) +func defaultOIDCRetryPolicy() oidcRetryPolicy { + p := oidcRetryPolicy{ + Attempts: 4, + PerAttemptTimeout: 20 * time.Second, + BaseBackoff: 1 * time.Second, + MaxBackoff: 8 * time.Second, + } + if v := os.Getenv("SC_OIDC_TOKEN_REQUEST_ATTEMPTS"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + p.Attempts = n + } + } + if v := os.Getenv("SC_OIDC_TOKEN_REQUEST_TIMEOUT"); v != "" { + if d, err := time.ParseDuration(v); err == nil && d > 0 { + p.PerAttemptTimeout = d } + } + return p +} - var tokenResp struct { - Value string `json:"value"` +func requestOIDCTokenWithRetry(ctx context.Context, requestURL, requestToken string, policy oidcRetryPolicy, sleep func(context.Context, time.Duration) error) (string, error) { + var lastErr error + for attempt := 0; attempt < policy.Attempts; attempt++ { + if err := ctx.Err(); err != nil { + return "", err } - if err := json.Unmarshal(body, &tokenResp); err != nil { - return fmt.Errorf("parsing OIDC token response: %w", err) + if attempt > 0 { + if err := sleep(ctx, oidcBackoff(policy, attempt)); err != nil { + return "", err + } } - if tokenResp.Value == "" { - return fmt.Errorf("OIDC token response has empty value field") + token, retryable, err := doOIDCTokenRequest(ctx, requestURL, requestToken, policy.PerAttemptTimeout) + if err == nil { + return token, nil + } + lastErr = err + if !retryable { + return "", err + } + if err := ctx.Err(); err != nil { + return "", err } - e.OIDCToken = tokenResp.Value - return nil } + return "", fmt.Errorf("requesting OIDC token failed after %d attempts: %w", policy.Attempts, lastErr) +} - return fmt.Errorf("OIDC token not available") +func doOIDCTokenRequest(ctx context.Context, requestURL, requestToken string, timeout time.Duration) (string, bool, error) { + attemptCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + reqURL, err := oidcTokenURL(requestURL) + if err != nil { + return "", false, err + } + req, err := http.NewRequestWithContext(attemptCtx, http.MethodGet, reqURL, nil) + if err != nil { + return "", false, fmt.Errorf("creating token request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+requestToken) + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return "", true, fmt.Errorf("requesting OIDC token: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return "", retryableOIDCStatus(resp.StatusCode), fmt.Errorf("OIDC token request failed with status %d: %s", resp.StatusCode, string(body)) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", true, fmt.Errorf("reading token response: %w", err) + } + var tokenResp struct { + Value string `json:"value"` + } + if err := json.Unmarshal(body, &tokenResp); err != nil { + return "", false, fmt.Errorf("parsing OIDC token response: %w", err) + } + if tokenResp.Value == "" { + return "", false, fmt.Errorf("OIDC token response has empty value field") + } + return tokenResp.Value, false, nil +} + +func retryableOIDCStatus(status int) bool { + if status == http.StatusRequestTimeout || status == http.StatusTooManyRequests { + return true + } + return status >= 500 && status <= 599 +} + +func oidcTokenURL(requestURL string) (string, error) { + u, err := url.Parse(requestURL) + if err != nil { + return "", fmt.Errorf("invalid OIDC request URL: %w", err) + } + q := u.Query() + q.Set("audience", "sigstore") + u.RawQuery = q.Encode() + return u.String(), nil +} + +func oidcBackoff(policy oidcRetryPolicy, attempt int) time.Duration { + maxBackoff := policy.BaseBackoff << (attempt - 1) + if maxBackoff <= 0 || maxBackoff > policy.MaxBackoff { + maxBackoff = policy.MaxBackoff + } + if maxBackoff <= 0 { + return 0 + } + return time.Duration(rand.Int63n(int64(maxBackoff))) +} + +func sleepContext(ctx context.Context, d time.Duration) error { + if d <= 0 { + return nil + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return nil + } } // PopulateGitMetadata populates git-related metadata from environment diff --git a/pkg/security/context_oidc_retry_test.go b/pkg/security/context_oidc_retry_test.go new file mode 100644 index 00000000..11894a2e --- /dev/null +++ b/pkg/security/context_oidc_retry_test.go @@ -0,0 +1,190 @@ +package security + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + . "github.com/onsi/gomega" +) + +type oidcResp struct { + status int + body string + delay time.Duration +} + +func noSleep(context.Context, time.Duration) error { return nil } + +func fastPolicy(attempts int) oidcRetryPolicy { + return oidcRetryPolicy{ + Attempts: attempts, + PerAttemptTimeout: 2 * time.Second, + BaseBackoff: time.Millisecond, + MaxBackoff: time.Millisecond, + } +} + +func oidcServer(t *testing.T, responses []oidcResp) (*httptest.Server, *int32, chan string) { + t.Helper() + var hits int32 + urls := make(chan string, 16) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + urls <- r.URL.String() + i := int(atomic.AddInt32(&hits, 1)) - 1 + resp := responses[len(responses)-1] + if i < len(responses) { + resp = responses[i] + } + if resp.delay > 0 { + time.Sleep(resp.delay) + } + w.WriteHeader(resp.status) + _, _ = w.Write([]byte(resp.body)) + })) + t.Cleanup(srv.Close) + return srv, &hits, urls +} + +func TestRequestOIDCTokenSuccessFirstTry(t *testing.T) { + RegisterTestingT(t) + srv, hits, urls := oidcServer(t, []oidcResp{{status: 200, body: `{"value":"tok"}`}}) + + token, err := requestOIDCTokenWithRetry(context.Background(), srv.URL, "req-token", fastPolicy(4), noSleep) + Expect(err).ToNot(HaveOccurred()) + Expect(token).To(Equal("tok")) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(1))) + Expect(<-urls).To(ContainSubstring("audience=sigstore")) +} + +func TestRequestOIDCTokenRetriesOn5xxThenSucceeds(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 500}, {status: 502}, {status: 200, body: `{"value":"tok"}`}}) + + token, err := requestOIDCTokenWithRetry(context.Background(), srv.URL, "req-token", fastPolicy(4), noSleep) + Expect(err).ToNot(HaveOccurred()) + Expect(token).To(Equal("tok")) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(3))) +} + +func TestRequestOIDCTokenRetriesOn429(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 429}, {status: 200, body: `{"value":"tok"}`}}) + + token, err := requestOIDCTokenWithRetry(context.Background(), srv.URL, "req-token", fastPolicy(4), noSleep) + Expect(err).ToNot(HaveOccurred()) + Expect(token).To(Equal("tok")) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(2))) +} + +func TestRequestOIDCTokenFailsFastOn401(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 401, body: "bad credentials"}, {status: 200, body: `{"value":"tok"}`}}) + + _, err := requestOIDCTokenWithRetry(context.Background(), srv.URL, "req-token", fastPolicy(4), noSleep) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("401")) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(1))) +} + +func TestRequestOIDCTokenExhaustsOnPersistent503(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 503}}) + + _, err := requestOIDCTokenWithRetry(context.Background(), srv.URL, "req-token", fastPolicy(4), noSleep) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("after 4 attempts")) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(4))) +} + +func TestRequestOIDCTokenNoRetryOnMalformed200(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 200, body: "not-json"}}) + + _, err := requestOIDCTokenWithRetry(context.Background(), srv.URL, "req-token", fastPolicy(4), noSleep) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("parsing OIDC token response")) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(1))) +} + +func TestRequestOIDCTokenNoRetryOnEmptyValue(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 200, body: `{"value":""}`}}) + + _, err := requestOIDCTokenWithRetry(context.Background(), srv.URL, "req-token", fastPolicy(4), noSleep) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("empty value")) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(1))) +} + +func TestRequestOIDCTokenAbortsOnParentCancelDuringBackoff(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 500}, {status: 200, body: `{"value":"tok"}`}}) + + cancelSleep := func(context.Context, time.Duration) error { return context.Canceled } + _, err := requestOIDCTokenWithRetry(context.Background(), srv.URL, "req-token", fastPolicy(4), cancelSleep) + Expect(err).To(MatchError(context.Canceled)) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(1))) +} + +func TestRequestOIDCTokenAbortsWhenParentContextDone(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 200, body: `{"value":"tok"}`}}) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := requestOIDCTokenWithRetry(ctx, srv.URL, "req-token", fastPolicy(4), noSleep) + Expect(err).To(MatchError(context.Canceled)) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(0))) +} + +func TestRequestOIDCTokenRetriesOnPerAttemptTimeout(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 200, body: `{"value":"tok"}`, delay: 200 * time.Millisecond}}) + policy := oidcRetryPolicy{Attempts: 2, PerAttemptTimeout: 20 * time.Millisecond, BaseBackoff: time.Millisecond, MaxBackoff: time.Millisecond} + + _, err := requestOIDCTokenWithRetry(context.Background(), srv.URL, "req-token", policy, noSleep) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("after 2 attempts")) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(2))) +} + +func TestRetryableOIDCStatus(t *testing.T) { + RegisterTestingT(t) + for status, want := range map[int]bool{ + 408: true, 429: true, 500: true, 503: true, 599: true, + 200: false, 400: false, 401: false, 403: false, 404: false, + } { + Expect(retryableOIDCStatus(status)).To(Equal(want), "status %d", status) + } +} + +func TestOIDCTokenURLAddsAudienceAndPreservesParams(t *testing.T) { + RegisterTestingT(t) + got, err := oidcTokenURL("https://token.actions.example.com/req?api-version=2.0") + Expect(err).ToNot(HaveOccurred()) + Expect(got).To(ContainSubstring("audience=sigstore")) + Expect(got).To(ContainSubstring("api-version=2.0")) + + _, err = oidcTokenURL("http://example.com/%zz") + Expect(err).To(HaveOccurred()) +} + +func TestGetOIDCTokenGitHubActionsHappyPath(t *testing.T) { + RegisterTestingT(t) + srv, hits, _ := oidcServer(t, []oidcResp{{status: 200, body: `{"value":"ci-tok"}`}}) + clearCIEnv(t) + t.Setenv("SIGSTORE_ID_TOKEN", "") + t.Setenv("GITHUB_ACTIONS", "true") + t.Setenv("ACTIONS_ID_TOKEN_REQUEST_URL", srv.URL) + t.Setenv("ACTIONS_ID_TOKEN_REQUEST_TOKEN", "req-token") + + e := &ExecutionContext{} + e.DetectCI() + Expect(e.GetOIDCToken(context.Background())).ToNot(HaveOccurred()) + Expect(e.OIDCToken).To(Equal("ci-tok")) + Expect(atomic.LoadInt32(hits)).To(Equal(int32(1))) +} From 0f706e6a3f7abfe60f74febbb48eb3d48a271ace Mon Sep 17 00:00:00 2001 From: Dmitrii Creed Date: Wed, 17 Jun 2026 12:11:58 +0400 Subject: [PATCH 2/2] feat(k8s): support cloudExtras.topologySpreadConstraints Add a topologySpreadConstraints field to cloudExtras so a stack can spread its replicas across nodes/zones. On GKE Autopilot this is the cost-correct way to achieve node-level HA: pod anti-affinity forces a 0.5 vCPU minimum request per pod (Autopilot rejects pods below it), while topology spread is not subject to that minimum, so replicas spread at their normal request. The constraint is threaded through both cloudExtras->DeploymentConfig converters (kubernetes-native and GKE Autopilot) and into the pod spec. Conversion is split into a pure normalize step (defaults + validation: maxSkew>=1 default 1, whenUnsatisfiable default DoNotSchedule, minDomains >=1 and only with DoNotSchedule, labelSelector defaults to the deployment's own pods) and a mechanical Pulumi mapping, so the defaulting/validation is unit tested without resolving Pulumi outputs. Threading guards cover both converters since they copy cloudExtras field-by-field and have drifted apart before. To force a second node on Autopilot use whenUnsatisfiable: DoNotSchedule with minDomains: 2; ScheduleAnyway only expresses a preference and lets the scheduler bin-pack replicas onto one node. Signed-off-by: Dmitrii Creed --- .../examples/kubernetes-affinity/README.md | 35 ++++++ pkg/clouds/gcloud/gke_autopilot.go | 1 + pkg/clouds/gcloud/topology_spread_test.go | 31 +++++ pkg/clouds/k8s/kube_run.go | 12 ++ pkg/clouds/k8s/topology_spread_test.go | 67 +++++++++++ pkg/clouds/k8s/types.go | 2 + pkg/clouds/pulumi/kubernetes/deployment.go | 69 +++++------ .../pulumi/kubernetes/simple_container.go | 107 +++++++++++++++--- .../pulumi/kubernetes/topology_spread_test.go | 99 ++++++++++++++++ 9 files changed, 371 insertions(+), 52 deletions(-) create mode 100644 pkg/clouds/gcloud/topology_spread_test.go create mode 100644 pkg/clouds/k8s/topology_spread_test.go create mode 100644 pkg/clouds/pulumi/kubernetes/topology_spread_test.go diff --git a/docs/docs/examples/kubernetes-affinity/README.md b/docs/docs/examples/kubernetes-affinity/README.md index df90099b..9ab46682 100644 --- a/docs/docs/examples/kubernetes-affinity/README.md +++ b/docs/docs/examples/kubernetes-affinity/README.md @@ -76,6 +76,34 @@ stacks: topologyKey: "topology.kubernetes.io/zone" ``` +### **Spread Replicas Across Nodes (Topology Spread)** + +To keep a multi-replica service available across node failures, spread its replicas +across nodes. On **GKE Autopilot** prefer `topologySpreadConstraints` over +`podAntiAffinity`: pod anti-affinity forces a 0.5 vCPU minimum request per pod (and is +*rejected* outright below it), whereas topology spread is not subject to that minimum. + +```yaml +stacks: + my-service: + type: cloud-compose + config: + scale: + min: 2 + max: 4 + cloudExtras: + topologySpreadConstraints: + - topologyKey: "kubernetes.io/hostname" + maxSkew: 1 + whenUnsatisfiable: "DoNotSchedule" + minDomains: 2 +``` + +`labelSelector` defaults to the deployment's own pods when omitted. On GKE Autopilot, +`whenUnsatisfiable: DoNotSchedule` together with `minDomains: 2` is required to actually +force a second node — `ScheduleAnyway` only expresses a preference and lets the scheduler +bin-pack both replicas onto one node. + ## **Supported Affinity Properties** ### **Simple Container Properties** @@ -88,6 +116,13 @@ stacks: - **`podAffinity`**: Pod co-location rules - **`podAntiAffinity`**: Pod separation and distribution rules +### **Topology Spread Properties (`cloudExtras.topologySpreadConstraints`)** +- **`topologyKey`** (required): Node label defining a domain (e.g. `kubernetes.io/hostname`) +- **`maxSkew`**: Max allowed pod-count difference between domains (default `1`) +- **`whenUnsatisfiable`**: `DoNotSchedule` (default) or `ScheduleAnyway` +- **`minDomains`**: Minimum eligible domains (only valid with `DoNotSchedule`) +- **`labelSelector`**: Pods to count; defaults to this deployment's pods when omitted + ## **Implementation Details** ### **GKE Integration** diff --git a/pkg/clouds/gcloud/gke_autopilot.go b/pkg/clouds/gcloud/gke_autopilot.go index 82211d03..f4e5defa 100644 --- a/pkg/clouds/gcloud/gke_autopilot.go +++ b/pkg/clouds/gcloud/gke_autopilot.go @@ -109,6 +109,7 @@ func ToGkeAutopilotConfig(tpl any, composeCfg compose.Config, stackCfg *api.Stac deployCfg.LivenessProbe = k8sCloudExtras.LivenessProbe // Extract global liveness probe configuration deployCfg.StartupProbe = k8sCloudExtras.StartupProbe // Extract global startup probe configuration deployCfg.PriorityClassName = k8sCloudExtras.PriorityClassName // Extract PriorityClass for pod scheduling and preemption + deployCfg.TopologySpreadConstraints = k8sCloudExtras.TopologySpreadConstraints // Process affinity rules and merge with existing NodeSelector if needed if k8sCloudExtras.Affinity != nil { diff --git a/pkg/clouds/gcloud/topology_spread_test.go b/pkg/clouds/gcloud/topology_spread_test.go new file mode 100644 index 00000000..745b5318 --- /dev/null +++ b/pkg/clouds/gcloud/topology_spread_test.go @@ -0,0 +1,31 @@ +package gcloud + +import ( + "testing" + + "github.com/compose-spec/compose-go/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/simple-container-com/api/pkg/api" + "github.com/simple-container-com/api/pkg/clouds/compose" +) + +func TestTopologySpreadThreadedByToGkeAutopilotConfig(t *testing.T) { + cloudExtras := any(map[string]any{ + "topologySpreadConstraints": []map[string]any{ + {"topologyKey": "kubernetes.io/hostname", "minDomains": 2}, + }, + }) + stackCfg := &api.StackConfigCompose{ + Runs: []string{}, + CloudExtras: &cloudExtras, + } + + res, err := ToGkeAutopilotConfig(&GkeAutopilotTemplate{}, compose.Config{Project: &types.Project{}}, stackCfg) + require.NoError(t, err) + input, ok := res.(*GkeAutopilotInput) + require.True(t, ok) + require.Len(t, input.Deployment.TopologySpreadConstraints, 1) + assert.Equal(t, "kubernetes.io/hostname", input.Deployment.TopologySpreadConstraints[0].TopologyKey) +} diff --git a/pkg/clouds/k8s/kube_run.go b/pkg/clouds/k8s/kube_run.go index e0fc94b6..e3ef9a46 100644 --- a/pkg/clouds/k8s/kube_run.go +++ b/pkg/clouds/k8s/kube_run.go @@ -28,6 +28,17 @@ type CloudExtras struct { StartupProbe *CloudRunProbe `json:"startupProbe" yaml:"startupProbe"` EphemeralVolumes []GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage PriorityClassName *string `json:"priorityClassName" yaml:"priorityClassName"` // Kubernetes PriorityClass for pod scheduling and preemption + + TopologySpreadConstraints []TopologySpreadConstraint `json:"topologySpreadConstraints" yaml:"topologySpreadConstraints"` +} + +// TopologySpreadConstraint spreads pods across nodes without the GKE Autopilot 0.5 vCPU minimum that pod anti-affinity requires. +type TopologySpreadConstraint struct { + MaxSkew *int `json:"maxSkew" yaml:"maxSkew"` + TopologyKey string `json:"topologyKey" yaml:"topologyKey"` + WhenUnsatisfiable string `json:"whenUnsatisfiable" yaml:"whenUnsatisfiable"` + LabelSelector *LabelSelector `json:"labelSelector" yaml:"labelSelector"` + MinDomains *int `json:"minDomains" yaml:"minDomains"` } // AffinityRules defines pod affinity and anti-affinity rules for node pool isolation @@ -176,6 +187,7 @@ func ToKubernetesRunConfig(tpl any, composeCfg compose.Config, stackCfg *api.Sta deployCfg.LivenessProbe = k8sCloudExtras.LivenessProbe // Extract global liveness probe configuration deployCfg.EphemeralVolumes = k8sCloudExtras.EphemeralVolumes // Extract generic ephemeral volumes configuration deployCfg.PriorityClassName = k8sCloudExtras.PriorityClassName // Extract PriorityClass for pod scheduling and preemption + deployCfg.TopologySpreadConstraints = k8sCloudExtras.TopologySpreadConstraints // Process affinity rules and merge with existing NodeSelector if needed if k8sCloudExtras.Affinity != nil { diff --git a/pkg/clouds/k8s/topology_spread_test.go b/pkg/clouds/k8s/topology_spread_test.go new file mode 100644 index 00000000..d23a5977 --- /dev/null +++ b/pkg/clouds/k8s/topology_spread_test.go @@ -0,0 +1,67 @@ +package k8s + +import ( + "testing" + + "github.com/compose-spec/compose-go/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/simple-container-com/api/pkg/api" + "github.com/simple-container-com/api/pkg/clouds/compose" +) + +func TestTopologySpreadConstraintsExtraction(t *testing.T) { + cloudExtras := map[string]any{ + "topologySpreadConstraints": []map[string]any{ + { + "maxSkew": 1, + "topologyKey": "kubernetes.io/hostname", + "whenUnsatisfiable": "DoNotSchedule", + "minDomains": 2, + "labelSelector": map[string]any{ + "matchLabels": map[string]any{"tier": "web"}, + }, + }, + }, + } + + result, err := api.ConvertDescriptor(cloudExtras, &CloudExtras{}) + + require.NoError(t, err) + require.Len(t, result.TopologySpreadConstraints, 1) + c := result.TopologySpreadConstraints[0] + assert.Equal(t, "kubernetes.io/hostname", c.TopologyKey) + assert.Equal(t, "DoNotSchedule", c.WhenUnsatisfiable) + require.NotNil(t, c.MaxSkew) + assert.Equal(t, 1, *c.MaxSkew) + require.NotNil(t, c.MinDomains) + assert.Equal(t, 2, *c.MinDomains) + require.NotNil(t, c.LabelSelector) + assert.Equal(t, map[string]string{"tier": "web"}, c.LabelSelector.MatchLabels) +} + +func TestTopologySpreadConstraintsAbsent(t *testing.T) { + result, err := api.ConvertDescriptor(map[string]any{}, &CloudExtras{}) + require.NoError(t, err) + assert.Nil(t, result.TopologySpreadConstraints) +} + +func TestTopologySpreadThreadedByToKubernetesRunConfig(t *testing.T) { + cloudExtras := any(map[string]any{ + "topologySpreadConstraints": []map[string]any{ + {"topologyKey": "kubernetes.io/hostname"}, + }, + }) + stackCfg := &api.StackConfigCompose{ + Runs: []string{}, + CloudExtras: &cloudExtras, + } + + res, err := ToKubernetesRunConfig(&CloudrunTemplate{}, compose.Config{Project: &types.Project{}}, stackCfg) + require.NoError(t, err) + input, ok := res.(*KubeRunInput) + require.True(t, ok) + require.Len(t, input.Deployment.TopologySpreadConstraints, 1) + assert.Equal(t, "kubernetes.io/hostname", input.Deployment.TopologySpreadConstraints[0].TopologyKey) +} diff --git a/pkg/clouds/k8s/types.go b/pkg/clouds/k8s/types.go index 04fdf675..e905bd33 100644 --- a/pkg/clouds/k8s/types.go +++ b/pkg/clouds/k8s/types.go @@ -33,6 +33,8 @@ type DeploymentConfig struct { StartupProbe *CloudRunProbe `json:"startupProbe" yaml:"startupProbe"` // Global startup probe configuration EphemeralVolumes []GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage PriorityClassName *string `json:"priorityClassName" yaml:"priorityClassName"` // Kubernetes PriorityClass for pod scheduling and preemption + + TopologySpreadConstraints []TopologySpreadConstraint `json:"topologySpreadConstraints" yaml:"topologySpreadConstraints"` } type CaddyConfig struct { diff --git a/pkg/clouds/pulumi/kubernetes/deployment.go b/pkg/clouds/pulumi/kubernetes/deployment.go index bd7a239d..ebe73b12 100644 --- a/pkg/clouds/pulumi/kubernetes/deployment.go +++ b/pkg/clouds/pulumi/kubernetes/deployment.go @@ -227,40 +227,41 @@ func DeploySimpleContainer(ctx *sdk.Context, args Args, opts ...sdk.ResourceOpti args.Params.Log.Warn(ctx.Context(), "configure simple container deployment for %q in %q", stackName, stackEnv) sc, err := NewSimpleContainer(ctx, &SimpleContainerArgs{ - KubeProvider: args.KubeProvider, - ComputeContext: args.ComputeContext, - ServiceType: args.ServiceType, - ExternalTrafficPolicy: args.ExternalTrafficPolicy, - UseSSL: args.UseSSL, - ProvisionIngress: args.ProvisionIngress, - Namespace: namespace, - Service: deploymentName, - Deployment: deploymentName, - ScEnv: stackEnv, - IngressContainer: args.Deployment.IngressContainer, - Domain: lo.FromPtr(args.Deployment.StackConfig).Domain, - Prefix: lo.FromPtr(args.Deployment.StackConfig).Prefix, - ProxyKeepPrefix: lo.FromPtr(args.Deployment.StackConfig).ProxyKeepPrefix, - ParentStack: lo.If(args.Params.ParentStack != nil, lo.ToPtr(lo.FromPtr(args.Params.ParentStack).FullReference)).Else(nil), - ParentEnv: lo.If(parentEnv != "", lo.ToPtr(parentEnv)).Else(nil), - Replicas: replicas, - Headers: args.Deployment.Headers, - SecretEnvs: mergedSecretEnvs, - LbConfig: args.Deployment.StackConfig.LBConfig, - Volumes: args.Deployment.TextVolumes, - PersistentVolumes: pvs, - EphemeralVolumes: args.Deployment.EphemeralVolumes, // Pass generic ephemeral volumes configuration - PriorityClassName: args.Deployment.PriorityClassName, - Containers: containers, - ServiceAccountName: args.ServiceAccountName, - InitContainers: args.InitContainers, - GenerateCaddyfileEntry: args.GenerateCaddyfileEntry, - Annotations: args.Annotations, - NodeSelector: args.NodeSelector, - Affinity: args.Affinity, - Sidecars: args.Sidecars, - VPA: args.VPA, // Pass VPA configuration to SimpleContainer - Scale: args.Deployment.Scale, // Pass Scale configuration to SimpleContainer + KubeProvider: args.KubeProvider, + ComputeContext: args.ComputeContext, + ServiceType: args.ServiceType, + ExternalTrafficPolicy: args.ExternalTrafficPolicy, + UseSSL: args.UseSSL, + ProvisionIngress: args.ProvisionIngress, + Namespace: namespace, + Service: deploymentName, + Deployment: deploymentName, + ScEnv: stackEnv, + IngressContainer: args.Deployment.IngressContainer, + Domain: lo.FromPtr(args.Deployment.StackConfig).Domain, + Prefix: lo.FromPtr(args.Deployment.StackConfig).Prefix, + ProxyKeepPrefix: lo.FromPtr(args.Deployment.StackConfig).ProxyKeepPrefix, + ParentStack: lo.If(args.Params.ParentStack != nil, lo.ToPtr(lo.FromPtr(args.Params.ParentStack).FullReference)).Else(nil), + ParentEnv: lo.If(parentEnv != "", lo.ToPtr(parentEnv)).Else(nil), + Replicas: replicas, + Headers: args.Deployment.Headers, + SecretEnvs: mergedSecretEnvs, + LbConfig: args.Deployment.StackConfig.LBConfig, + Volumes: args.Deployment.TextVolumes, + PersistentVolumes: pvs, + EphemeralVolumes: args.Deployment.EphemeralVolumes, // Pass generic ephemeral volumes configuration + PriorityClassName: args.Deployment.PriorityClassName, + Containers: containers, + ServiceAccountName: args.ServiceAccountName, + InitContainers: args.InitContainers, + GenerateCaddyfileEntry: args.GenerateCaddyfileEntry, + Annotations: args.Annotations, + NodeSelector: args.NodeSelector, + Affinity: args.Affinity, + TopologySpreadConstraints: args.Deployment.TopologySpreadConstraints, + Sidecars: args.Sidecars, + VPA: args.VPA, // Pass VPA configuration to SimpleContainer + Scale: args.Deployment.Scale, // Pass Scale configuration to SimpleContainer PodDisruption: lo.If(args.Deployment.DisruptionBudget != nil, args.Deployment.DisruptionBudget).Else(&k8s.DisruptionBudget{ MinAvailable: lo.ToPtr(1), }), diff --git a/pkg/clouds/pulumi/kubernetes/simple_container.go b/pkg/clouds/pulumi/kubernetes/simple_container.go index 918212be..2954d223 100644 --- a/pkg/clouds/pulumi/kubernetes/simple_container.go +++ b/pkg/clouds/pulumi/kubernetes/simple_container.go @@ -106,24 +106,25 @@ type SimpleContainerArgs struct { KubeProvider sdk.ProviderResource // optional properties - PodDisruption *k8s.DisruptionBudget `json:"podDisruption" yaml:"podDisruption"` - LbConfig *api.SimpleContainerLBConfig `json:"lbConfig" yaml:"lbConfig"` - SecretEnvs map[string]string `json:"secretEnvs" yaml:"secretEnvs"` - Annotations map[string]string `json:"annotations" yaml:"annotations"` - NodeSelector map[string]string `json:"nodeSelector" yaml:"nodeSelector"` - Affinity *k8s.AffinityRules `json:"affinity" yaml:"affinity"` - PriorityClassName *string `json:"priorityClassName" yaml:"priorityClassName"` // Kubernetes PriorityClass for pod scheduling and preemption - IngressContainer *k8s.CloudRunContainer `json:"ingressContainer" yaml:"ingressContainer"` - ServiceType *string `json:"serviceType" yaml:"serviceType"` - ExternalTrafficPolicy *string `json:"externalTrafficPolicy" yaml:"externalTrafficPolicy"` - ProvisionIngress bool `json:"provisionIngress" yaml:"provisionIngress"` - Headers *k8s.Headers `json:"headers" yaml:"headers"` - Volumes []k8s.SimpleTextVolume `json:"volumes" yaml:"volumes"` - SecretVolumes []k8s.SimpleTextVolume `json:"secretVolumes" yaml:"secretVolumes"` - PersistentVolumes []k8s.PersistentVolume `json:"persistentVolumes" yaml:"persistentVolumes"` - EphemeralVolumes []k8s.GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage - VPA *k8s.VPAConfig `json:"vpa" yaml:"vpa"` - Scale *k8s.Scale `json:"scale" yaml:"scale"` + PodDisruption *k8s.DisruptionBudget `json:"podDisruption" yaml:"podDisruption"` + LbConfig *api.SimpleContainerLBConfig `json:"lbConfig" yaml:"lbConfig"` + SecretEnvs map[string]string `json:"secretEnvs" yaml:"secretEnvs"` + Annotations map[string]string `json:"annotations" yaml:"annotations"` + NodeSelector map[string]string `json:"nodeSelector" yaml:"nodeSelector"` + Affinity *k8s.AffinityRules `json:"affinity" yaml:"affinity"` + TopologySpreadConstraints []k8s.TopologySpreadConstraint `json:"topologySpreadConstraints" yaml:"topologySpreadConstraints"` + PriorityClassName *string `json:"priorityClassName" yaml:"priorityClassName"` // Kubernetes PriorityClass for pod scheduling and preemption + IngressContainer *k8s.CloudRunContainer `json:"ingressContainer" yaml:"ingressContainer"` + ServiceType *string `json:"serviceType" yaml:"serviceType"` + ExternalTrafficPolicy *string `json:"externalTrafficPolicy" yaml:"externalTrafficPolicy"` + ProvisionIngress bool `json:"provisionIngress" yaml:"provisionIngress"` + Headers *k8s.Headers `json:"headers" yaml:"headers"` + Volumes []k8s.SimpleTextVolume `json:"volumes" yaml:"volumes"` + SecretVolumes []k8s.SimpleTextVolume `json:"secretVolumes" yaml:"secretVolumes"` + PersistentVolumes []k8s.PersistentVolume `json:"persistentVolumes" yaml:"persistentVolumes"` + EphemeralVolumes []k8s.GenericEphemeralVolume `json:"ephemeralVolumes" yaml:"ephemeralVolumes"` // Generic ephemeral volumes for large temp storage + VPA *k8s.VPAConfig `json:"vpa" yaml:"vpa"` + Scale *k8s.Scale `json:"scale" yaml:"scale"` Log logger.Logger // ... @@ -594,6 +595,12 @@ func NewSimpleContainer(ctx *sdk.Context, args *SimpleContainerArgs, opts ...sdk convertedAffinity := convertAffinityRulesToKubernetes(args.Affinity) args.Log.Info(ctx.Context(), "🔍 DEBUG: Converted affinity result: %+v", convertedAffinity) + normalizedTSC, tscErr := normalizeTopologySpreadConstraints(args.TopologySpreadConstraints, appLabels) + if tscErr != nil { + return nil, tscErr + } + topologySpread := convertTopologySpreadConstraints(normalizedTSC) + podSpecArgs := &corev1.PodSpecArgs{ NodeSelector: sdk.ToStringMap(args.NodeSelector), Affinity: convertedAffinity, @@ -629,6 +636,9 @@ func NewSimpleContainer(ctx *sdk.Context, args *SimpleContainerArgs, opts ...sdk if args.PriorityClassName != nil { podSpecArgs.PriorityClassName = sdk.String(*args.PriorityClassName) } + if topologySpread != nil { + podSpecArgs.TopologySpreadConstraints = topologySpread + } if imagePullSecret != nil { podSpecArgs.ImagePullSecrets = corev1.LocalObjectReferenceArray{ corev1.LocalObjectReferenceArgs{ @@ -1427,3 +1437,64 @@ func convertLabelSelector(labelSelector *k8s.LabelSelector) *metav1.LabelSelecto return kubeLabelSelector } + +func normalizeTopologySpreadConstraints(constraints []k8s.TopologySpreadConstraint, appLabels map[string]string) ([]k8s.TopologySpreadConstraint, error) { + if len(constraints) == 0 { + return nil, nil + } + + out := make([]k8s.TopologySpreadConstraint, 0, len(constraints)) + for i, c := range constraints { + if c.TopologyKey == "" { + return nil, errors.Errorf("topologySpreadConstraints[%d]: topologyKey is required", i) + } + + n := c + if n.WhenUnsatisfiable == "" { + n.WhenUnsatisfiable = "DoNotSchedule" + } + if n.WhenUnsatisfiable != "DoNotSchedule" && n.WhenUnsatisfiable != "ScheduleAnyway" { + return nil, errors.Errorf("topologySpreadConstraints[%d]: whenUnsatisfiable must be DoNotSchedule or ScheduleAnyway, got %q", i, n.WhenUnsatisfiable) + } + if n.MaxSkew == nil { + n.MaxSkew = lo.ToPtr(1) + } else if *n.MaxSkew < 1 { + return nil, errors.Errorf("topologySpreadConstraints[%d]: maxSkew must be >= 1, got %d", i, *n.MaxSkew) + } + if n.MinDomains != nil { + if *n.MinDomains < 1 { + return nil, errors.Errorf("topologySpreadConstraints[%d]: minDomains must be >= 1, got %d", i, *n.MinDomains) + } + if n.WhenUnsatisfiable != "DoNotSchedule" { + return nil, errors.Errorf("topologySpreadConstraints[%d]: minDomains requires whenUnsatisfiable=DoNotSchedule", i) + } + } + if n.LabelSelector == nil { + n.LabelSelector = &k8s.LabelSelector{MatchLabels: appLabels} + } + + out = append(out, n) + } + return out, nil +} + +func convertTopologySpreadConstraints(constraints []k8s.TopologySpreadConstraint) corev1.TopologySpreadConstraintArray { + if len(constraints) == 0 { + return nil + } + + result := make(corev1.TopologySpreadConstraintArray, 0, len(constraints)) + for _, c := range constraints { + constraint := corev1.TopologySpreadConstraintArgs{ + MaxSkew: sdk.Int(lo.FromPtr(c.MaxSkew)), + TopologyKey: sdk.String(c.TopologyKey), + WhenUnsatisfiable: sdk.String(c.WhenUnsatisfiable), + LabelSelector: convertLabelSelector(c.LabelSelector), + } + if c.MinDomains != nil { + constraint.MinDomains = sdk.Int(*c.MinDomains) + } + result = append(result, constraint) + } + return result +} diff --git a/pkg/clouds/pulumi/kubernetes/topology_spread_test.go b/pkg/clouds/pulumi/kubernetes/topology_spread_test.go new file mode 100644 index 00000000..d1051224 --- /dev/null +++ b/pkg/clouds/pulumi/kubernetes/topology_spread_test.go @@ -0,0 +1,99 @@ +package kubernetes + +import ( + "testing" + + "github.com/samber/lo" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/simple-container-com/api/pkg/clouds/k8s" +) + +func TestNormalizeTopologySpreadConstraintsEmpty(t *testing.T) { + out, err := normalizeTopologySpreadConstraints(nil, map[string]string{"app": "x"}) + require.NoError(t, err) + assert.Nil(t, out) +} + +func TestNormalizeTopologySpreadConstraintsDefaults(t *testing.T) { + appLabels := map[string]string{"app-name": "web", "sc-env": "prod"} + out, err := normalizeTopologySpreadConstraints([]k8s.TopologySpreadConstraint{ + {TopologyKey: "kubernetes.io/hostname"}, + }, appLabels) + require.NoError(t, err) + require.Len(t, out, 1) + assert.Equal(t, 1, lo.FromPtr(out[0].MaxSkew)) + assert.Equal(t, "DoNotSchedule", out[0].WhenUnsatisfiable) + require.NotNil(t, out[0].LabelSelector) + assert.Equal(t, appLabels, out[0].LabelSelector.MatchLabels) +} + +func TestNormalizeTopologySpreadConstraintsExplicitValuesPreserved(t *testing.T) { + sel := &k8s.LabelSelector{MatchLabels: map[string]string{"tier": "web"}} + out, err := normalizeTopologySpreadConstraints([]k8s.TopologySpreadConstraint{ + {TopologyKey: "topology.kubernetes.io/zone", MaxSkew: lo.ToPtr(3), WhenUnsatisfiable: "ScheduleAnyway", LabelSelector: sel}, + }, map[string]string{"app": "x"}) + require.NoError(t, err) + require.Len(t, out, 1) + assert.Equal(t, 3, lo.FromPtr(out[0].MaxSkew)) + assert.Equal(t, "ScheduleAnyway", out[0].WhenUnsatisfiable) + assert.Equal(t, map[string]string{"tier": "web"}, out[0].LabelSelector.MatchLabels) +} + +func TestNormalizeTopologySpreadConstraintsMinDomains(t *testing.T) { + out, err := normalizeTopologySpreadConstraints([]k8s.TopologySpreadConstraint{ + {TopologyKey: "kubernetes.io/hostname", MinDomains: lo.ToPtr(2)}, + }, map[string]string{"app": "x"}) + require.NoError(t, err) + require.Len(t, out, 1) + assert.Equal(t, 2, lo.FromPtr(out[0].MinDomains)) +} + +func TestNormalizeTopologySpreadConstraintsErrors(t *testing.T) { + cases := map[string]k8s.TopologySpreadConstraint{ + "empty topologyKey": {TopologyKey: ""}, + "invalid whenUnsatisfiable": {TopologyKey: "kubernetes.io/hostname", WhenUnsatisfiable: "Maybe"}, + "maxSkew zero": {TopologyKey: "kubernetes.io/hostname", MaxSkew: lo.ToPtr(0)}, + "maxSkew negative": {TopologyKey: "kubernetes.io/hostname", MaxSkew: lo.ToPtr(-2)}, + "minDomains without DoNotSched": {TopologyKey: "kubernetes.io/hostname", WhenUnsatisfiable: "ScheduleAnyway", MinDomains: lo.ToPtr(2)}, + "minDomains zero": {TopologyKey: "kubernetes.io/hostname", MinDomains: lo.ToPtr(0)}, + "minDomains negative": {TopologyKey: "kubernetes.io/hostname", MinDomains: lo.ToPtr(-1)}, + } + for name, c := range cases { + t.Run(name, func(t *testing.T) { + _, err := normalizeTopologySpreadConstraints([]k8s.TopologySpreadConstraint{c}, map[string]string{"app": "x"}) + assert.Error(t, err) + }) + } +} + +func TestNormalizeTopologySpreadConstraintsDoesNotMutateInput(t *testing.T) { + in := []k8s.TopologySpreadConstraint{{TopologyKey: "kubernetes.io/hostname"}} + _, err := normalizeTopologySpreadConstraints(in, map[string]string{"app": "x"}) + require.NoError(t, err) + assert.Nil(t, in[0].MaxSkew) + assert.Equal(t, "", in[0].WhenUnsatisfiable) + assert.Nil(t, in[0].LabelSelector) +} + +func TestNormalizeTopologySpreadConstraintsMultiple(t *testing.T) { + out, err := normalizeTopologySpreadConstraints([]k8s.TopologySpreadConstraint{ + {TopologyKey: "kubernetes.io/hostname"}, + {TopologyKey: "topology.kubernetes.io/zone", WhenUnsatisfiable: "ScheduleAnyway"}, + }, map[string]string{"app": "x"}) + require.NoError(t, err) + require.Len(t, out, 2) + assert.Equal(t, "kubernetes.io/hostname", out[0].TopologyKey) + assert.Equal(t, "topology.kubernetes.io/zone", out[1].TopologyKey) +} + +func TestConvertTopologySpreadConstraints(t *testing.T) { + assert.Nil(t, convertTopologySpreadConstraints(nil)) + + normalized := []k8s.TopologySpreadConstraint{ + {TopologyKey: "kubernetes.io/hostname", MaxSkew: lo.ToPtr(1), WhenUnsatisfiable: "DoNotSchedule", MinDomains: lo.ToPtr(2), LabelSelector: &k8s.LabelSelector{MatchLabels: map[string]string{"app": "x"}}}, + } + out := convertTopologySpreadConstraints(normalized) + require.Len(t, out, 1) +}