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) +} 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))) +}