diff --git a/pkg/admin/local.go b/pkg/admin/local.go index cf429e0..268a4c1 100644 --- a/pkg/admin/local.go +++ b/pkg/admin/local.go @@ -117,7 +117,7 @@ func (l *adminService) CreateBridge(ctx context.Context, req *bridgev1.CreateBri bundle, err = resources.SourceFromManifests(req.SourceManifests) } else if req.SourceDeployment != "" { logger.Info("Creating bridge") - bundle, err = resources.SourceFromNamespace(ctx, l.client, targetNS, req.SourceDeployment) + bundle, err = resources.SourceFromNamespace(ctx, l.dynClient, targetNS, req.SourceDeployment) } else { logger.Info("Creating simple bridge") bundle, err = resources.SourceSimple(targetNS, proxyImage) diff --git a/pkg/commands/administrator.go b/pkg/commands/administrator.go index a53cbd0..a61166c 100644 --- a/pkg/commands/administrator.go +++ b/pkg/commands/administrator.go @@ -203,7 +203,8 @@ func (s *administratorServer) DeleteBridge(ctx context.Context, req *bridgev1.De // gRPC status error. func grpcError(err error) error { var notFound *resources.DeploymentNotFoundError - if errors.As(err, ¬Found) { + var workloadNotFound *resources.WorkloadNotFoundError + if errors.As(err, ¬Found) || errors.As(err, &workloadNotFound) { return status.Error(codes.NotFound, err.Error()) } return status.Error(codes.Internal, err.Error()) diff --git a/pkg/k8s/resources/resources.go b/pkg/k8s/resources/resources.go index b50ce3b..2ee5fa3 100644 --- a/pkg/k8s/resources/resources.go +++ b/pkg/k8s/resources/resources.go @@ -32,6 +32,17 @@ const ( DefaultWorkloadKind = "Deployment" ) +// WorkloadNotFoundError is returned when no workload of any supported kind has +// the source name. +type WorkloadNotFoundError struct { + Name string + Namespace string +} + +func (e *WorkloadNotFoundError) Error() string { + return fmt.Sprintf("no workload found named '%s' in namespace '%s'", e.Name, e.Namespace) +} + // DeploymentNotFoundError is returned when the source deployment does not exist. type DeploymentNotFoundError struct { Name string diff --git a/pkg/k8s/resources/source.go b/pkg/k8s/resources/source.go index be0f0d7..7141d1d 100644 --- a/pkg/k8s/resources/source.go +++ b/pkg/k8s/resources/source.go @@ -8,27 +8,34 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/client-go/kubernetes" + "k8s.io/client-go/dynamic" ) -// SourceFromNamespace fetches a source deployment from the cluster and returns -// a Bundle containing just that deployment. Only the deployment spec is used — -// not the live pod — so that webhook-injected env vars and volume mounts (e.g. -// IRSA) are absent and get cleanly re-injected on the bridge pod. -func SourceFromNamespace(ctx context.Context, client kubernetes.Interface, namespace, deployment string) (*Bundle, error) { - srcDeploy, err := client.AppsV1().Deployments(namespace).Get(ctx, deployment, metav1.GetOptions{}) - if err != nil { +// SourceFromNamespace fetches the named workload from the cluster, trying each +// of workloadKinds in order, and returns a Bundle holding it as a Deployment +// (see DeploymentFromWorkload). Only the workload's pod template is used — not a +// live pod — so that webhook-injected env vars and volume mounts (e.g. IRSA) +// are absent and get cleanly re-injected on the bridge pod. +func SourceFromNamespace(ctx context.Context, client dynamic.Interface, namespace, name string) (*Bundle, error) { + for _, kind := range workloadKinds { + obj, err := client.Resource(gvkToGVR(kind.GVK())).Namespace(namespace).Get(ctx, name, metav1.GetOptions{}) if errors.IsNotFound(err) { - return nil, &DeploymentNotFoundError{Name: deployment, Namespace: namespace} + continue } - return nil, fmt.Errorf("failed to get source deployment %s/%s: %w", namespace, deployment, err) + if err != nil { + return nil, fmt.Errorf("failed to get source %s %s/%s: %w", kind.GVK().Kind, namespace, name, err) + } + deploy, err := DeploymentFromWorkload(kind, obj) + if err != nil { + return nil, err + } + return &Bundle{ + Resources: []Resource{ + {Object: deploy, GVK: appsv1.SchemeGroupVersion.WithKind("Deployment")}, + }, + }, nil } - - return &Bundle{ - Resources: []Resource{ - {Object: srcDeploy, GVK: appsv1.SchemeGroupVersion.WithKind("Deployment")}, - }, - }, nil + return nil, &WorkloadNotFoundError{Name: name, Namespace: namespace} } // SourceSimple builds a minimal Deployment with just the bridge proxy container. diff --git a/pkg/k8s/resources/source_manifests.go b/pkg/k8s/resources/source_manifests.go index 63c2fa5..56264a5 100644 --- a/pkg/k8s/resources/source_manifests.go +++ b/pkg/k8s/resources/source_manifests.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "log/slog" + "slices" "strings" "github.com/vercel/bridge/pkg/archive" @@ -20,7 +21,8 @@ import ( // SourceFromManifests unpacks a tar.gz archive of Kubernetes YAML files, parses // every document into typed or unstructured resources, and returns a Bundle -// ready for the transform pipeline. +// ready for the transform pipeline. Every workload, whatever its kind, becomes a +// Deployment (see DeploymentFromWorkload), ordered by workloadKinds. func SourceFromManifests(manifests []byte) (*Bundle, error) { fileMap, err := archive.Unpack(manifests) if err != nil { @@ -28,6 +30,11 @@ func SourceFromManifests(manifests []byte) (*Bundle, error) { } var resources []Resource + type workload struct { + rank int + deploy *appsv1.Deployment + } + var workloads []workload decoder := scheme.Codecs.UniversalDeserializer() @@ -38,6 +45,17 @@ func SourceFromManifests(manifests []byte) (*Bundle, error) { continue } + if u, err := decodeUnstructured(doc); err == nil { + if i := workloadKindIndex(u.GroupVersionKind()); i >= 0 { + deploy, err := DeploymentFromWorkload(workloadKinds[i], u) + if err != nil { + return nil, fmt.Errorf("%s: %w", name, err) + } + workloads = append(workloads, workload{rank: i, deploy: deploy}) + continue + } + } + obj, gvk, err := decoder.Decode(doc, nil, nil) if err != nil { // Try as unstructured for unknown types. @@ -51,8 +69,6 @@ func SourceFromManifests(manifests []byte) (*Bundle, error) { } switch obj.(type) { - case *appsv1.Deployment: - resources = append(resources, Resource{Object: obj, GVK: *gvk}) case *corev1.Service, *corev1.ConfigMap, *corev1.Secret, *corev1.ServiceAccount: resources = append(resources, Resource{Object: obj, GVK: *gvk}) default: @@ -65,6 +81,11 @@ func SourceFromManifests(manifests []byte) (*Bundle, error) { } } + slices.SortStableFunc(workloads, func(a, b workload) int { return a.rank - b.rank }) + for _, w := range workloads { + resources = append(resources, Resource{Object: w.deploy, GVK: appsv1.SchemeGroupVersion.WithKind("Deployment")}) + } + return &Bundle{ Resources: resources, }, nil diff --git a/pkg/k8s/resources/workload.go b/pkg/k8s/resources/workload.go new file mode 100644 index 0000000..f37dde8 --- /dev/null +++ b/pkg/k8s/resources/workload.go @@ -0,0 +1,128 @@ +package resources + +import ( + "fmt" + "maps" + "strings" + + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +// WorkloadKind is a kind of workload a bridge can be created from. +type WorkloadKind interface { + // GVK is the workload's group, version and kind. + GVK() schema.GroupVersionKind + // PodTemplatePath is the field path of the workload's pod template. + PodTemplatePath() []string +} + +type deploymentKind struct{} + +func (*deploymentKind) GVK() schema.GroupVersionKind { + return appsv1.SchemeGroupVersion.WithKind("Deployment") +} +func (*deploymentKind) PodTemplatePath() []string { return []string{"spec", "template"} } + +type statefulSetKind struct{} + +func (*statefulSetKind) GVK() schema.GroupVersionKind { + return appsv1.SchemeGroupVersion.WithKind("StatefulSet") +} +func (*statefulSetKind) PodTemplatePath() []string { return []string{"spec", "template"} } + +type daemonSetKind struct{} + +func (*daemonSetKind) GVK() schema.GroupVersionKind { + return appsv1.SchemeGroupVersion.WithKind("DaemonSet") +} +func (*daemonSetKind) PodTemplatePath() []string { return []string{"spec", "template"} } + +type replicaSetKind struct{} + +func (*replicaSetKind) GVK() schema.GroupVersionKind { + return appsv1.SchemeGroupVersion.WithKind("ReplicaSet") +} +func (*replicaSetKind) PodTemplatePath() []string { return []string{"spec", "template"} } + +type jobKind struct{} + +func (*jobKind) GVK() schema.GroupVersionKind { + return batchv1.SchemeGroupVersion.WithKind("Job") +} +func (*jobKind) PodTemplatePath() []string { return []string{"spec", "template"} } + +type cronJobKind struct{} + +func (*cronJobKind) GVK() schema.GroupVersionKind { + return batchv1.SchemeGroupVersion.WithKind("CronJob") +} +func (*cronJobKind) PodTemplatePath() []string { + return []string{"spec", "jobTemplate", "spec", "template"} +} + +// workloadKinds are the workload kinds a bridge can be created from, in the +// order a name is looked up and a bundle's workloads are ordered. +var workloadKinds = []WorkloadKind{ + &deploymentKind{}, + &statefulSetKind{}, + &daemonSetKind{}, + &replicaSetKind{}, + &jobKind{}, + &cronJobKind{}, +} + +// workloadKindIndex returns gvk's position in workloadKinds, or -1 if it isn't a +// workload kind. +func workloadKindIndex(gvk schema.GroupVersionKind) int { + for i, kind := range workloadKinds { + if k := kind.GVK(); k.Group == gvk.Group && k.Kind == gvk.Kind { + return i + } + } + return -1 +} + +// DeploymentFromWorkload returns a single-replica Deployment built from a +// workload's pod template and metadata, so the rest of the pipeline only ever +// deals with Deployments. +// +// A bridge built from a Job or CronJob runs its pod, not its schedule or +// completion: run the job in the devcontainer with `bridge exec`. +func DeploymentFromWorkload(kind WorkloadKind, obj *unstructured.Unstructured) (*appsv1.Deployment, error) { + path := kind.PodTemplatePath() + raw, found, err := unstructured.NestedMap(obj.Object, path...) + if err != nil || !found { + return nil, fmt.Errorf("%s %s has no pod template at %s", kind.GVK().Kind, obj.GetName(), strings.Join(path, ".")) + } + var template corev1.PodTemplateSpec + if err := runtime.DefaultUnstructuredConverter.FromUnstructured(raw, &template); err != nil { + return nil, fmt.Errorf("invalid pod template in %s %s: %w", kind.GVK().Kind, obj.GetName(), err) + } + // A Deployment's pods must restart Always and can't set a deadline. + template.Spec.RestartPolicy = corev1.RestartPolicyAlways + template.Spec.ActiveDeadlineSeconds = nil + if len(template.Labels) == 0 { + template.Labels = map[string]string{"app": obj.GetName()} + } + + replicas := int32(1) + return &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: obj.GetName(), + Namespace: obj.GetNamespace(), + Labels: obj.GetLabels(), + Annotations: obj.GetAnnotations(), + }, + Spec: appsv1.DeploymentSpec{ + Replicas: &replicas, + Selector: &metav1.LabelSelector{MatchLabels: maps.Clone(template.Labels)}, + Template: template, + }, + }, nil +} diff --git a/pkg/k8s/resources/workload_test.go b/pkg/k8s/resources/workload_test.go new file mode 100644 index 0000000..59b60dc --- /dev/null +++ b/pkg/k8s/resources/workload_test.go @@ -0,0 +1,222 @@ +package resources + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + dynamicfake "k8s.io/client-go/dynamic/fake" + k8stesting "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/kubernetes/scheme" +) + +// podTemplateYAML is a pod template, indented to sit under a `template:` key +// at the given depth. +func podTemplateYAML(indent string) string { + return indent + `metadata: +` + indent + ` labels: +` + indent + ` app: my-workload +` + indent + `spec: +` + indent + ` serviceAccountName: my-sa +` + indent + ` restartPolicy: Never +` + indent + ` activeDeadlineSeconds: 600 +` + indent + ` containers: +` + indent + ` - name: main +` + indent + ` image: my-workload:latest +` + indent + ` env: +` + indent + ` - name: FOO +` + indent + ` value: bar +` +} + +const workloadMetadata = `metadata: + name: my-workload + namespace: jobs + labels: + team: platform + annotations: + owner: me +` + +func workloadYAML(apiVersion, kind string) string { + head := "apiVersion: " + apiVersion + "\nkind: " + kind + "\n" + workloadMetadata + if kind == "CronJob" { + return head + "spec:\n schedule: \"0 * * * *\"\n jobTemplate:\n spec:\n template:\n" + podTemplateYAML(" ") + } + return head + "spec:\n template:\n" + podTemplateYAML(" ") +} + +func TestDeploymentFromWorkload_EveryKind(t *testing.T) { + for _, kind := range workloadKinds { + gvk := kind.GVK() + t.Run(gvk.Kind, func(t *testing.T) { + obj, err := decodeUnstructured([]byte(workloadYAML(gvk.GroupVersion().String(), gvk.Kind))) + require.NoError(t, err) + + deploy, err := DeploymentFromWorkload(kind, obj) + require.NoError(t, err) + require.NotNil(t, deploy) + + assert.Equal(t, "my-workload", deploy.Name) + assert.Equal(t, "jobs", deploy.Namespace) + assert.Equal(t, map[string]string{"team": "platform"}, deploy.Labels) + assert.Equal(t, map[string]string{"owner": "me"}, deploy.Annotations) + require.NotNil(t, deploy.Spec.Replicas) + assert.Equal(t, int32(1), *deploy.Spec.Replicas) + assert.Equal(t, map[string]string{"app": "my-workload"}, deploy.Spec.Selector.MatchLabels) + + pod := deploy.Spec.Template.Spec + assert.Equal(t, "my-sa", pod.ServiceAccountName) + assert.Equal(t, corev1.RestartPolicyAlways, pod.RestartPolicy) + assert.Nil(t, pod.ActiveDeadlineSeconds) + require.Len(t, pod.Containers, 1) + assert.Contains(t, pod.Containers[0].Env, corev1.EnvVar{Name: "FOO", Value: "bar"}) + }) + } +} + +func TestDeploymentFromWorkload_NoPodTemplate(t *testing.T) { + obj, err := decodeUnstructured([]byte("apiVersion: batch/v1\nkind: CronJob\nmetadata:\n name: my-cron\nspec:\n schedule: \"0 * * * *\"\n")) + require.NoError(t, err) + + _, err = DeploymentFromWorkload(&cronJobKind{}, obj) + assert.ErrorContains(t, err, "no pod template at spec.jobTemplate.spec.template") +} + +func TestDeploymentFromWorkload_UnlabeledTemplate(t *testing.T) { + obj, err := decodeUnstructured([]byte(`apiVersion: batch/v1 +kind: Job +metadata: + name: my-job +spec: + template: + spec: + containers: + - name: main + image: my-job:latest +`)) + require.NoError(t, err) + + deploy, err := DeploymentFromWorkload(&jobKind{}, obj) + require.NoError(t, err) + assert.Equal(t, map[string]string{"app": "my-job"}, deploy.Spec.Template.Labels) + assert.Equal(t, map[string]string{"app": "my-job"}, deploy.Spec.Selector.MatchLabels) +} + +func TestCreateFromManifests_CronJob(t *testing.T) { + manifests := packTestManifests(t, map[string]string{ + "cron.yaml": workloadYAML("batch/v1", "CronJob"), + }) + + client := k8stesting.NewSimpleClientset() + deployName, bundle, err := testCreateFromManifests(t, client, manifests, "default") + require.NoError(t, err) + assert.Equal(t, testDeployName("my-workload"), deployName) + + for _, r := range bundle.Resources { + _, isCronJob := r.Object.(*batchv1.CronJob) + assert.False(t, isCronJob, "the CronJob must be bridged, not saved as a CronJob") + } + + deploy, err := client.AppsV1().Deployments("default").Get(context.Background(), deployName, metav1.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, "my-sa", deploy.Spec.Template.Spec.ServiceAccountName) + assert.Equal(t, corev1.RestartPolicyAlways, deploy.Spec.Template.Spec.RestartPolicy) +} + +func TestSourceFromManifests_DeploymentIsTheSource(t *testing.T) { + manifests := packTestManifests(t, map[string]string{ + "cron.yaml": workloadYAML("batch/v1", "CronJob"), + "config.yaml": "apiVersion: v1\nkind: ConfigMap\nmetadata:\n name: config\ndata:\n a: b\n", + "deploy.yaml": `apiVersion: apps/v1 +kind: Deployment +metadata: + name: my-app +spec: + selector: + matchLabels: + app: my-app + template: + metadata: + labels: + app: my-app + spec: + containers: + - name: app + image: my-app:latest +`, + }) + + bundle, err := SourceFromManifests(manifests) + require.NoError(t, err) + assert.Equal(t, "my-app", FindDeploymentName(bundle)) + + var deployments []string + var configMaps int + for _, r := range bundle.Resources { + switch o := r.Object.(type) { + case *appsv1.Deployment: + deployments = append(deployments, o.Name) + case *corev1.ConfigMap: + configMaps++ + } + } + assert.Equal(t, []string{"my-app", "my-workload"}, deployments) + assert.Equal(t, 1, configMaps, "a ConfigMap is kept as a ConfigMap") +} + +func TestSourceFromNamespace(t *testing.T) { + template := corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "my-workload"}}, + Spec: corev1.PodSpec{ + ServiceAccountName: "cron-sa", + Containers: []corev1.Container{{Name: "main", Image: "my-workload:latest"}}, + }, + } + cronJob := &batchv1.CronJob{ + ObjectMeta: metav1.ObjectMeta{Name: "my-workload", Namespace: "default"}, + Spec: batchv1.CronJobSpec{JobTemplate: batchv1.JobTemplateSpec{ + Spec: batchv1.JobSpec{Template: template}, + }}, + } + + t.Run("finds a CronJob", func(t *testing.T) { + client := dynamicfake.NewSimpleDynamicClient(scheme.Scheme, cronJob) + + bundle, err := SourceFromNamespace(context.Background(), client, "default", "my-workload") + require.NoError(t, err) + require.Len(t, bundle.Resources, 1) + deploy, ok := bundle.Resources[0].Object.(*appsv1.Deployment) + require.True(t, ok) + assert.Equal(t, "my-workload", deploy.Name) + assert.Equal(t, "cron-sa", deploy.Spec.Template.Spec.ServiceAccountName) + }) + + t.Run("prefers a Deployment of the same name", func(t *testing.T) { + deployTemplate := *template.DeepCopy() + deployTemplate.Spec.ServiceAccountName = "deploy-sa" + client := dynamicfake.NewSimpleDynamicClient(scheme.Scheme, cronJob, &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: "my-workload", Namespace: "default"}, + Spec: appsv1.DeploymentSpec{Template: deployTemplate}, + }) + + bundle, err := SourceFromNamespace(context.Background(), client, "default", "my-workload") + require.NoError(t, err) + deploy := bundle.Resources[0].Object.(*appsv1.Deployment) + assert.Equal(t, "deploy-sa", deploy.Spec.Template.Spec.ServiceAccountName) + }) + + t.Run("is not found when no workload has the name", func(t *testing.T) { + client := dynamicfake.NewSimpleDynamicClient(scheme.Scheme) + + _, err := SourceFromNamespace(context.Background(), client, "default", "missing") + var notFound *WorkloadNotFoundError + assert.ErrorAs(t, err, ¬Found) + }) +}