Kubernetes Operators in Go: Building an AWS EC2 Controller

package v1

import (
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// Ec2InstanceSpec defines the desired state of Ec2Instance.
type Ec2InstanceSpec struct {
  // instanceType is the EC2 virtual machine sizing (e.g., t3.micro, t3.small).
  // +kubebuilder:validation:Required
  // +kubebuilder:validation:Enum=t2.micro;t3.nano;t3.micro;t3.small;t3.medium
  // +kubebuilder:default="t3.micro"
  InstanceType string `json:"instanceType"`

  // amiId is the Amazon Machine Image identifier.
  // +kubebuilder:validation:Required
  // +kubebuilder:validation:Pattern="^ami-[0-9a-f]{8,17}$"
  AMIId string `json:"amiId"`

  // region specifies the AWS target region (e.g. us-east-1, eu-west-1).
  // +kubebuilder:validation:Required
  Region string `json:"region"`

  // keyPair is the optional AWS EC2 KeyPair name for SSH access.
  // +optional
  KeyPair string `json:"keyPair,omitempty"`

  // securityGroups is an optional list of Security Group IDs.
  // +optional
  SecurityGroups []string `json:"securityGroups,omitempty"`

  // subnet specifies the AWS VPC Subnet ID.
  // +optional
  Subnet string `json:"subnet,omitempty"`

  // tags are custom AWS tags applied to the provisioned instance.
  // +optional
  Tags map[string]string `json:"tags,omitempty"`
}

// Ec2InstanceStatus defines the observed state of Ec2Instance.
type Ec2InstanceStatus struct {
  // instanceId is the AWS EC2 instance identifier assigned upon launch.
  // +optional
  InstanceID string `json:"instanceId,omitempty"`

  // state is the current AWS instance lifecycle state (pending, running, stopped, terminated).
  // +optional
  State string `json:"state,omitempty"`

  // publicIP is the allocated IPv4 public address.
  // +optional
  PublicIP string `json:"publicIP,omitempty"`

  // privateIP is the internal VPC IPv4 address.
  // +optional
  PrivateIP string `json:"privateIP,omitempty"`

  // conditions track lifecycle milestones for status monitoring.
  // +listType=map
  // +listMapKey=type
  // +optional
  Conditions []metav1.Condition `json:"conditions,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:printcolumn:name="InstanceType",type="string",JSONPath=".spec.instanceType"
// +kubebuilder:printcolumn:name="State",type="string",JSONPath=".status.state"
// +kubebuilder:printcolumn:name="PublicIP",type="string",JSONPath=".status.publicIP"
// +kubebuilder:printcolumn:name="InstanceID",type="string",JSONPath=".status.instanceId"
// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
type Ec2Instance struct {
  metav1.TypeMeta   `json:",inline"`
  metav1.ObjectMeta `json:"metadata,omitzero"`

  Spec   Ec2InstanceSpec   `json:"spec,omitempty"`
  Status Ec2InstanceStatus `json:"status,omitzero"`
}

// +kubebuilder:object:root=true
type Ec2InstanceList struct {
  metav1.TypeMeta `json:",inline"`
  metav1.ListMeta `json:"metadata,omitempty"`
  Items           []Ec2Instance `json:"items"`
}

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: ec2instances.compute.cloud.com
spec:
  group: compute.cloud.com
  names:
    kind: Ec2Instance
    plural: ec2instances
  scope: Namespaced
  versions:
  - name: v1
    served: true
    storage: true
    subresources:
      status: {}
    additionalPrinterColumns:
    - name: InstanceType
      type: string
      jsonPath: .spec.instanceType
    - name: State
      type: string
      jsonPath: .status.state
    - name: PublicIP
      type: string
      jsonPath: .status.publicIP
    - name: InstanceID
      type: string
      jsonPath: .status.instanceId
    - name: Age
      type: date
      jsonPath: .metadata.creationTimestamp

package v1

import (
  "context"
  "fmt"

  "k8s.io/apimachinery/pkg/runtime"
  ctrl "sigs.k8s.io/controller-runtime"
  logf "sigs.k8s.io/controller-runtime/pkg/log"
  "sigs.k8s.io/controller-runtime/pkg/webhook"
)

var ec2log = logf.Log.WithName("ec2instance-resource")

func (r *Ec2Instance) SetupWebhookWithManager(mgr ctrl.Manager) error {
  return ctrl.NewWebhookManagedBy(mgr).
    For(r).
    Complete()
}

// +kubebuilder:webhook:path=/mutate-compute-cloud-com-v1-ec2instance,mutating=true,failurePolicy=fail,sideEffects=None,groups=compute.cloud.com,resources=ec2instances,verbs=create;update,versions=v1,name=mec2instance.kb.io,admissionReviewVersions=v1

var _ webhook.Defaulter = &Ec2Instance{}

// Default implements webhook.Defaulter so a webhook will be registered for the type.
func (r *Ec2Instance) Default() {
  ec2log.Info("defaulting Ec2Instance", "name", r.Name)

  // 1. Default AWS Region if omitted
  if r.Spec.Region == "" {
    r.Spec.Region = "us-east-1"
  }

  // 2. Default Instance Type
  if r.Spec.InstanceType == "" {
    r.Spec.InstanceType = "t3.micro"
  }

  // 3. Enforce corporate governance & cost tracking tags
  if r.Spec.Tags == nil {
    r.Spec.Tags = make(map[string]string)
  }
  if _, exists := r.Spec.Tags["ManagedBy"]; !exists {
    r.Spec.Tags["ManagedBy"] = "Kubernetes-Operator"
  }
  if _, exists := r.Spec.Tags["Environment"]; !exists {
    r.Spec.Tags["Environment"] = "Production"
  }
}

// +kubebuilder:webhook:path=/validate-compute-cloud-com-v1-ec2instance,mutating=false,failurePolicy=fail,sideEffects=None,groups=compute.cloud.com,resources=ec2instances,verbs=create;update;delete,versions=v1,name=vec2instance.kb.io,admissionReviewVersions=v1

var _ webhook.Validator = &Ec2Instance{}

// ValidateCreate validates the object upon creation.
func (r *Ec2Instance) ValidateCreate() (admission.Warnings, error) {
  ec2log.Info("validate create", "name", r.Name)
  return nil, r.validateEc2Instance()
}

func (r *Ec2Instance) validateEc2Instance() error {
  // Cross-field checks (e.g. subnet vs public IP)
  return nil
}

// ValidateUpdate enforces field immutability when an existing CR is edited.
func (r *Ec2Instance) ValidateUpdate(old runtime.Object) (admission.Warnings, error) {
  ec2log.Info("validate update", "name", r.Name)
  oldInstance, ok := old.(*Ec2Instance)
  if !ok {
    return nil, apierrors.NewInternalError(fmt.Errorf("expected an Ec2Instance but got %T", old))
  }

  var allErrs field.ErrorList

  // IMMUTABILITY CHECK: AWS EC2 Region cannot be modified in-place!
  if r.Spec.Region != oldInstance.Spec.Region {
    allErrs = append(allErrs, field.Forbidden(
      field.NewPath("spec").Child("region"),
      "region is immutable; recreate the Ec2Instance to deploy in a different AWS region",
    ))
  }

  // IMMUTABILITY CHECK: AMI ID cannot be modified on a running instance
  if r.Spec.AMIId != oldInstance.Spec.AMIId {
    allErrs = append(allErrs, field.Forbidden(
      field.NewPath("spec").Child("amiId"),
      "amiId is immutable; re-provision the resource to boot a new base image",
    ))
  }

  if len(allErrs) == 0 {
    return nil, nil
  }
  return nil, apierrors.NewInvalid(
    r.GroupVersionKind().GroupKind(),
    r.Name,
    allErrs,
  )
}

// ValidateDelete guards against accidental deletion of production instances.
func (r *Ec2Instance) ValidateDelete() (admission.Warnings, error) {
  ec2log.Info("validate delete", "name", r.Name)

  if r.Spec.Tags["Environment"] == "Production" {
    if r.Annotations == nil || r.Annotations["compute.cloud.com/safe-to-delete"] != "true" {
      return nil, apierrors.NewForbidden(
        schema.GroupResource{Group: "compute.cloud.com", Resource: "ec2instances"},
        r.Name,
        fmt.Errorf("cannot delete production instance without annotation 'compute.cloud.com/safe-to-delete: true'"),
      )
    }
  }
  return nil, nil
}

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: validating-webhook-configuration
  annotations:
    # cert-manager injects the CA bundle here automatically:
    cert-manager.io/inject-ca-from: default/ec2-operator-serving-cert
webhooks:
- name: vec2instance.compute.cloud.com
  admissionReviewVersions: ["v1"]
  clientConfig:
    service:
      name: ec2-operator-webhook-service
      namespace: default
      path: /validate-compute-cloud-com-v1-ec2instance
  failurePolicy: Fail
  rules:
  - apiGroups: ["compute.cloud.com"]
    apiVersions: ["v1"]
    operations: ["CREATE", "UPDATE", "DELETE"]
    resources: ["ec2instances"]
  sideEffects: None

package controller

import (
  "context"
  "time"

  "k8s.io/apimachinery/pkg/api/errors"
  "k8s.io/apimachinery/pkg/runtime"
  ctrl "sigs.k8s.io/controller-runtime"
  "sigs.k8s.io/controller-runtime/pkg/client"
  "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
  "sigs.k8s.io/controller-runtime/pkg/log"

  computev1 "github.com/r4rajat/ec2-operator/api/v1"
)

const ec2Finalizer = "ec2instance.compute.cloud.com/finalizer"

type Ec2InstanceReconciler struct {
  client.Client
  Scheme *runtime.Scheme
}

// +kubebuilder:rbac:groups=compute.cloud.com,resources=ec2instances,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=compute.cloud.com,resources=ec2instances/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=compute.cloud.com,resources=ec2instances/finalizers,verbs=update

func (r *Ec2InstanceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
  l := log.FromContext(ctx)

  // 1. Fetch current resource state from the local informer cache
  ec2Instance := &computev1.Ec2Instance{}
  if err := r.Get(ctx, req.NamespacedName, ec2Instance); err != nil {
    if errors.IsNotFound(err) {
      // Object was deleted from etcd; nothing more to reconcile
      return ctrl.Result{}, nil
    }
    return ctrl.Result{}, err
  }

  // 2. Check if the object is undergoing deletion
  if !ec2Instance.DeletionTimestamp.IsZero() {
    l.Info("Resource is marked for deletion. Running cloud cleanup...")
    if controllerutil.ContainsFinalizer(ec2Instance, ec2Finalizer) {
      // Terminate the AWS EC2 instance
      if _, err := deleteEc2Instance(ctx, ec2Instance); err != nil {
        l.Error(err, "Failed to terminate EC2 instance in AWS")
        return ctrl.Result{Requeue: true}, err
      }

      // Cloud resource confirmed terminated: strip finalizer
      controllerutil.RemoveFinalizer(ec2Instance, ec2Finalizer)
      if err := r.Update(ctx, ec2Instance); err != nil {
        return ctrl.Result{Requeue: true}, err
      }
    }
    return ctrl.Result{}, nil
  }

  // 3. Register finalizer BEFORE creating any cloud resources
  if !controllerutil.ContainsFinalizer(ec2Instance, ec2Finalizer) {
    controllerutil.AddFinalizer(ec2Instance, ec2Finalizer)
    if err := r.Update(ctx, ec2Instance); err != nil {
      return ctrl.Result{Requeue: true}, err
    }
    // r.Update emits a watch event that queues Reconcile #2,
    // but Reconcile #1 continues executing sequentially.
  }

  // 4. If AWS instance already exists: run Drift Detection
  if ec2Instance.Status.InstanceID != "" {
    exists, inst, err := checkEC2InstanceExists(ctx, ec2Instance.Status.InstanceID, ec2Instance)
    if err != nil {
      return ctrl.Result{RequeueAfter: 15 * time.Second}, err
    }

    if exists && string(inst.State.Name) != ec2Instance.Status.State {
      l.Info("Drift detected in AWS instance state!", "awsState", inst.State.Name)
      ec2Instance.Status.State = string(inst.State.Name)
      if err := r.Status().Update(ctx, ec2Instance); err != nil {
        return ctrl.Result{}, err
      }
    }
    // Recheck cloud drift periodically every 30 seconds
    return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
  }

  // 5. Create new AWS EC2 instance
  l.Info("Creating new AWS EC2 instance...", "type", ec2Instance.Spec.InstanceType)
  createdInfo, err := createEc2Instance(ec2Instance)
  if err != nil {
    l.Error(err, "AWS RunInstances failed")
    return ctrl.Result{}, err
  }

  // 6. Update Status subresource with observed AWS state
  ec2Instance.Status.InstanceID = createdInfo.InstanceID
  ec2Instance.Status.State = createdInfo.State
  ec2Instance.Status.PublicIP = createdInfo.PublicIP
  ec2Instance.Status.PrivateIP = createdInfo.PrivateIP

  if err := r.Status().Update(ctx, ec2Instance); err != nil {
    l.Error(err, "Failed to update Ec2Instance status")
    return ctrl.Result{}, err
  }

  return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
}

package controller

import (
  "context"
  "fmt"
  "time"

  "github.com/aws/aws-sdk-go-v2/aws"
  "github.com/aws/aws-sdk-go-v2/config"
  "github.com/aws/aws-sdk-go-v2/service/ec2"
  ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
  computev1 "github.com/r4rajat/ec2-operator/api/v1"
)

func getAWSClient(ctx context.Context, region string) (*ec2.Client, error) {
  // Loads credentials from AWS IRSA, EKS Pod Identity, or environment variables
  cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region))
  if err != nil {
    return nil, fmt.Errorf("unable to load AWS SDK config: %w", err)
  }
  return ec2.NewFromConfig(cfg), nil
}

func createEc2Instance(cr *computev1.Ec2Instance) (*computev1.Ec2InstanceStatus, error) {
  client, err := getAWSClient(context.TODO(), cr.Spec.Region)
  if err != nil {
    return nil, err
  }

  runInput := &ec2.RunInstancesInput{
    ImageId:      aws.String(cr.Spec.AMIId),
    InstanceType: ec2types.InstanceType(cr.Spec.InstanceType),
    MinCount:     aws.Int32(1),
    MaxCount:     aws.Int32(1),
    TagSpecifications: []ec2types.TagSpecification{
      {
        ResourceType: ec2types.ResourceTypeInstance,
        Tags: []ec2types.Tag{
          {Key: aws.String("ManagedBy"), Value: aws.String("Kubernetes-Operator")},
          {Key: aws.String("CRName"), Value: aws.String(cr.Name)},
        },
      },
    },
  }

  out, err := client.RunInstances(context.TODO(), runInput)
  if err != nil {
    return nil, fmt.Errorf("RunInstances failed: %w", err)
  }

  instID := *out.Instances[0].InstanceId

  // Use AWS SDK v2 Waiter to block until instance enters 'running'
  waiter := ec2.NewInstanceRunningWaiter(client)
  err = waiter.Wait(context.TODO(), &ec2.DescribeInstancesInput{
    InstanceIds: []string{instID},
  }, 3*time.Minute)
  if err != nil {
    return nil, fmt.Errorf("timeout waiting for instance to run: %w", err)
  }

  // Describe instance to capture assigned public/private IP addresses
  desc, _ := client.DescribeInstances(context.TODO(), &ec2.DescribeInstancesInput{
    InstanceIds: []string{instID},
  })
  runningInst := desc.Reservations[0].Instances[0]

  return &computev1.Ec2InstanceStatus{
    InstanceID: instID,
    State:      string(runningInst.State.Name),
    PublicIP:   aws.ToString(runningInst.PublicIpAddress),
    PrivateIP:  aws.ToString(runningInst.PrivateIpAddress),
  }, nil
}

func checkEC2InstanceExists(ctx context.Context, instanceID string, cr *computev1.Ec2Instance) (bool, *ec2types.Instance, error) {
  client, err := getAWSClient(ctx, cr.Spec.Region)
  if err != nil {
    return false, nil, err
  }

  result, err := client.DescribeInstances(ctx, &ec2.DescribeInstancesInput{
    InstanceIds: []string{instanceID},
  })
  if err != nil {
    return false, nil, err
  }

  if len(result.Reservations) == 0 || len(result.Reservations[0].Instances) == 0 {
    return false, nil, nil
  }
  return true, &result.Reservations[0].Instances[0], nil
}

func deleteEc2Instance(ctx context.Context, cr *computev1.Ec2Instance) (bool, error) {
  if cr.Status.InstanceID == "" {
    return true, nil // No cloud instance ever launched
  }

  client, err := getAWSClient(ctx, cr.Spec.Region)
  if err != nil {
    return false, err
  }

  _, err = client.TerminateInstances(ctx, &ec2.TerminateInstancesInput{
    InstanceIds: []string{cr.Status.InstanceID},
  })
  if err != nil {
    return false, err
  }

  // Wait until AWS confirms instance is fully terminated
  termWaiter := ec2.NewInstanceTerminatedWaiter(client)
  err = termWaiter.Wait(ctx, &ec2.DescribeInstancesInput{
    InstanceIds: []string{cr.Status.InstanceID},
  }, 5*time.Minute)

  return err == nil, err
}

func (r *Ec2InstanceReconciler) SetupWithManager(mgr ctrl.Manager) error {
  return ctrl.NewControllerManagedBy(mgr).
    // Watch changes to primary Ec2Instance resources
    For(&computev1.Ec2Instance{}).
    // Filter out events that do not modify metadata.generation (e.g. status updates)
    WithEventFilter(predicate.GenerationChangedPredicate{}).
    Complete(r)
}

var _ = Describe("Ec2Instance controller", func() {
  Context("When creating Ec2Instance with invalid schema", func() {
    It("Should be rejected by the API server before reaching controller", func() {
      ctx := context.Background()
      invalid := &computev1.Ec2Instance{
        ObjectMeta: metav1.ObjectMeta{
          Name:      "bad-instance",
          Namespace: "default",
        },
        Spec: computev1.Ec2InstanceSpec{
          InstanceType: "unsupported.large", // violates enum!
          AMIId:        "invalid-ami-string",  // violates regex!
          Region:       "us-east-1",
        },
      }
      err := k8sClient.Create(ctx, invalid)
      Expect(err).To(HaveOccurred())
      Expect(errors.IsInvalid(err)).To(BeTrue())
    })
  })
})

apiVersion: compute.cloud.com/v1
kind: Ec2Instance
metadata:
  name: web-server-prod
  namespace: default
spec:
  instanceType: t3.micro
  amiId: ami-0c55b159cbfafe1f0
  region: us-east-1
  tags:
    Environment: Production
    ManagedBy: KubernetesOperator

func RunClientGoController(ctx context.Context, dynClient dynamic.Interface) {
  gvr := schema.GroupVersionResource{
    Group:    "compute.cloud.com",
    Version:  "v1",
    Resource: "ec2instances",
  }

  factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynClient, 0, "default", nil)
  informer := factory.ForResource(gvr).Informer()

  queue := workqueue.NewTypedRateLimitingQueue[string](
    workqueue.DefaultTypedControllerRateLimiter[string](),
  )

  // Enqueue ONLY the namespace/name key — never the object!
  informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc: func(obj interface{}) {
      key, _ := cache.MetaNamespaceKeyFunc(obj)
      queue.Add(key)
    },
    UpdateFunc: func(_, newObj interface{}) {
      key, _ := cache.MetaNamespaceKeyFunc(newObj)
      queue.Add(key)
    },
  })

  factory.Start(ctx.Done())
  cache.WaitForCacheSync(ctx.Done(), informer.HasSynced)

  // Worker processing loop
  for {
    key, shutdown := queue.Get()
    if shutdown { break }

    err := syncKey(ctx, key)
    if err != nil {
      queue.AddRateLimited(key) // retry with exponential backoff
    } else {
      queue.Forget(key)         // reset backoff on success
    }
    queue.Done(key)
  }
}