egress-proxies-operator/internal/provider/gcp/gcp.go

// Package gcp implements the provider contract on GCP Compute Engine via
// the modern Cloud Client Library (cloud.google.com/go/compute/apiv1),
// deliberately restricted to four calls: instances.Insert, Get, Delete,
// AggregatedList. Operations are fire-and-forget — Operation.Wait is never
// called; Create/Delete return as soon as the operation is submitted and
// the reconciler discovers progress by polling Get.
package gcp

import (
	"context"
	"fmt"
	"strings"
	"time"

	compute "cloud.google.com/go/compute/apiv1"
	"cloud.google.com/go/compute/apiv1/computepb"
	"github.com/go-logr/logr"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"
	"google.golang.org/protobuf/proto"
	logf "sigs.k8s.io/controller-runtime/pkg/log"

	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)

// instancesAPI is the test seam. It deliberately does not mirror the SDK:
// the SDK's InstancesScopedListPairIterator has an unexported nextFunc, so
// a fake cannot construct one — the seam flattens AggregatedList to a
// slice, and returns operations as just their name (the only thing this
// provider ever uses, since it never waits on them).
type instancesAPI interface {
	Insert(ctx context.Context, req *computepb.InsertInstanceRequest) (opName string, err error)
	Get(ctx context.Context, req *computepb.GetInstanceRequest) (*computepb.Instance, error)
	Delete(ctx context.Context, req *computepb.DeleteInstanceRequest) (opName string, err error)
	AggregatedList(ctx context.Context, req *computepb.AggregatedListInstancesRequest) ([]*computepb.Instance, error)
}

// realInstances adapts *compute.InstancesClient to the seam.
type realInstances struct {
	client *compute.InstancesClient
}

func (r *realInstances) Insert(ctx context.Context, req *computepb.InsertInstanceRequest) (string, error) {
	op, err := r.client.Insert(ctx, req)
	if err != nil {
		return "", err
	}
	return op.Name(), nil
}

func (r *realInstances) Get(ctx context.Context, req *computepb.GetInstanceRequest) (*computepb.Instance, error) {
	return r.client.Get(ctx, req)
}

func (r *realInstances) Delete(ctx context.Context, req *computepb.DeleteInstanceRequest) (string, error) {
	op, err := r.client.Delete(ctx, req)
	if err != nil {
		return "", err
	}
	return op.Name(), nil
}

func (r *realInstances) AggregatedList(ctx context.Context, req *computepb.AggregatedListInstancesRequest) ([]*computepb.Instance, error) {
	it := r.client.AggregatedList(ctx, req)
	var out []*computepb.Instance
	for {
		pair, err := it.Next()
		if err == iterator.Done {
			return out, nil
		}
		if err != nil {
			return nil, err
		}
		if pair.Value != nil {
			out = append(out, pair.Value.Instances...)
		}
	}
}

// Provider implements provider.Provider on GCP Compute Engine.
type Provider struct {
	name string
	cfg  provider.GCPConfig
	api  instancesAPI
}

// New builds a Provider using Application Default Credentials (workload
// identity in-cluster, gcloud ADC locally — no key-file plumbing).
// Deliberately untested: it dials real Google endpoints; everything below
// it is exercised through newWithAPI.
func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, error) {
	return NewWithWireOptions(ctx, pc, WireLogOptions{})
}

// NewWithWireOptions is New with explicit control over the V(5) wire
// logging; the injected wire logger surfaces the SDK's HTTP
// request/response records at V(5). Note option.WithLogger overrides the
// SDK's own GOOGLE_SDK_GO_LOGGING_LEVEL env var, so --zap-log-level is
// the only knob.
func NewWithWireOptions(ctx context.Context, pc provider.ProviderConfig, opts WireLogOptions) (provider.Provider, error) {
	base := logf.Log.WithName("gcp").WithName("http")
	if base.V(5).Enabled() {
		logf.Log.WithName("gcp").Info(
			"GCP HTTP wire logging active — request payloads include cloud-init user-data",
			"fullPayloads", opts.FullPayloads)
	}
	client, err := compute.NewInstancesRESTClient(ctx, option.WithLogger(wireLogger(base, opts)))
	if err != nil {
		return nil, fmt.Errorf("creating GCP instances client: %w", err)
	}
	return newWithAPI(pc, &realInstances{client: client}), nil
}

func newWithAPI(pc provider.ProviderConfig, api instancesAPI) *Provider {
	cfg := provider.GCPConfig{}
	if pc.GCP != nil {
		cfg = *pc.GCP
	}
	return &Provider{name: pc.Name, cfg: withDefaults(cfg), api: api}
}

// logger derives the request-scoped logger from ctx, so provider lines
// inherit the reconcile context (which Proxy triggered the call).
func (p *Provider) logger(ctx context.Context) logr.Logger {
	return logf.FromContext(ctx).WithName("gcp").WithValues("provider", p.name)
}

// Create submits the insert and returns immediately with the
// zone-qualified providerID. A 409 alreadyExists is success — the
// deterministic instance name means a repeat call after a crash found the
// VM it already created, which is exactly the idempotency the contract
// demands.
func (p *Provider) Create(ctx context.Context, req provider.CreateRequest) (string, error) {
	pl := req.Placement
	if pl.Zone == "" || pl.MachineType == "" || pl.Image == "" {
		return "", provider.Wrap(provider.ErrPermanent, "create", p.name, "", fmt.Errorf(
			"gcp requires placement.zone, placement.machineType and placement.image (got zone=%q machineType=%q image=%q)",
			pl.Zone, pl.MachineType, pl.Image))
	}
	log := p.logger(ctx)
	id := formatProviderID(pl.Zone, req.Name)
	insertReq := buildInsertRequest(p.cfg, req)
	// Curated fields only: the request proto embeds the cloud-init
	// user-data, which may be Secret-sourced and must never reach logs.
	log.V(2).Info("GCP insert request built",
		"zone", pl.Zone, "name", req.Name,
		"network", p.cfg.Network, "networkTag", p.cfg.NetworkTag,
		"diskSizeGb", p.cfg.DiskSizeGB, "port", req.Port,
		"labels", insertReq.GetInstanceResource().GetLabels(),
		"cloudInitBytes", len(req.CloudInit))
	opName, err := p.api.Insert(ctx, insertReq)
	switch {
	case err == nil:
		log.V(1).Info("GCP instance insert submitted",
			"zone", pl.Zone, "name", req.Name,
			"machineType", pl.MachineType, "image", pl.Image, "opName", opName)
	case isAlreadyExists(err):
		log.V(1).Info("GCP instance already exists, insert treated as success",
			"zone", pl.Zone, "name", req.Name)
	default:
		logAPIError(log, "create", err)
		return "", p.wrapErr("create", id, err)
	}
	return id, nil
}

// Get returns the instance state. The providerID carries its own zone, so
// this stays correct even mid-replacement after a zone edit — re-reading
// spec.placement.zone would look up the wrong zone exactly then.
func (p *Provider) Get(ctx context.Context, providerID string) (*provider.Instance, error) {
	zone, name, err := parseProviderID(providerID)
	if err != nil {
		return nil, provider.Wrap(provider.ErrPermanent, "get", p.name, providerID, err)
	}
	log := p.logger(ctx)
	inst, err := p.api.Get(ctx, &computepb.GetInstanceRequest{
		Project:  p.cfg.Project,
		Zone:     zone,
		Instance: name,
	})
	if err != nil {
		logAPIError(log, "get", err)
		return nil, p.wrapErr("get", providerID, err)
	}
	out := toInstance(inst, zone)
	log.V(1).Info("GCP instance fetched",
		"zone", zone, "name", name,
		"status", inst.GetStatus(), "state", out.State, "ip", out.IP)
	return out, nil
}

// Delete submits the delete and returns; deleting an instance that is
// already gone is success.
func (p *Provider) Delete(ctx context.Context, providerID string) error {
	zone, name, err := parseProviderID(providerID)
	if err != nil {
		return provider.Wrap(provider.ErrPermanent, "delete", p.name, providerID, err)
	}
	log := p.logger(ctx)
	opName, err := p.api.Delete(ctx, &computepb.DeleteInstanceRequest{
		Project:  p.cfg.Project,
		Zone:     zone,
		Instance: name,
	})
	switch {
	case err == nil:
		log.V(1).Info("GCP instance delete submitted",
			"zone", zone, "name", name, "opName", opName)
	case isNotFound(err):
		log.V(1).Info("GCP instance already gone, delete treated as success",
			"zone", zone, "name", name)
	default:
		logAPIError(log, "delete", err)
		return p.wrapErr("delete", providerID, err)
	}
	return nil
}

// ListByTag sweeps every zone for instances carrying the GC labels.
// ReturnPartialSuccess matters: without it one unreachable zone fails the
// entire GC sweep.
func (p *Provider) ListByTag(ctx context.Context) ([]provider.Instance, error) {
	log := p.logger(ctx)
	filter := fmt.Sprintf("labels.%s = %s", provider.LabelManaged, provider.LabelManagedYes)
	instances, err := p.api.AggregatedList(ctx, &computepb.AggregatedListInstancesRequest{
		Project:              p.cfg.Project,
		Filter:               proto.String(filter),
		ReturnPartialSuccess: proto.Bool(true),
	})
	if err != nil {
		logAPIError(log, "list", err)
		return nil, p.wrapErr("list", "", err)
	}
	log.V(1).Info("GCP instances listed", "filter", filter, "count", len(instances))
	out := make([]provider.Instance, 0, len(instances))
	for _, inst := range instances {
		conv := toInstance(inst, lastPathSegment(inst.GetZone()))
		out = append(out, *conv)
		log.V(2).Info("GCP listed instance",
			"id", conv.ID, "state", conv.State, "uid", conv.UID, "createdAt", conv.CreatedAt)
	}
	return out, nil
}

func toInstance(inst *computepb.Instance, zone string) *provider.Instance {
	var ip string
	if nics := inst.GetNetworkInterfaces(); len(nics) > 0 {
		if acs := nics[0].GetAccessConfigs(); len(acs) > 0 {
			ip = acs[0].GetNatIP()
		}
	}
	// CreationTimestamp is RFC3339; a parse failure leaves the zero time,
	// which orphan GC treats as "old" — safe, since a malformed timestamp
	// never protects a candidate from collection forever.
	created, _ := time.Parse(time.RFC3339, inst.GetCreationTimestamp())
	return &provider.Instance{
		ID:        formatProviderID(zone, inst.GetName()),
		IP:        ip,
		State:     mapState(inst.GetStatus(), ip),
		UID:       inst.GetLabels()[provider.LabelUID],
		CreatedAt: created,
	}
}

// mapState collapses GCP instance statuses onto the provider states. A
// RUNNING instance without a NatIP maps to Provisioning — an empty IP must
// never be published as Running. Anything unrecognized maps to Stopped:
// the reconciler's answer to Stopped is delete-and-recreate, which is
// always safe for cattle.
func mapState(status, ip string) provider.InstanceState {
	switch status {
	case "PROVISIONING", "STAGING", "REPAIRING":
		return provider.StateProvisioning
	case "RUNNING":
		if ip == "" {
			return provider.StateProvisioning
		}
		return provider.StateRunning
	case "STOPPING", "STOPPED", "SUSPENDING", "SUSPENDED":
		return provider.StateStopped
	case "TERMINATED":
		return provider.StateTerminated
	default:
		return provider.StateStopped
	}
}

func formatProviderID(zone, name string) string {
	return fmt.Sprintf("zones/%s/instances/%s", zone, name)
}

func parseProviderID(id string) (zone, name string, err error) {
	parts := strings.Split(id, "/")
	if len(parts) != 4 || parts[0] != "zones" || parts[2] != "instances" || parts[1] == "" || parts[3] == "" {
		return "", "", fmt.Errorf("malformed gcp providerID %q, want zones/<zone>/instances/<name>", id)
	}
	return parts[1], parts[3], nil
}

// lastPathSegment extracts the zone name from the URL-style
// ".../zones/europe-west1-b" the API returns on instances.
func lastPathSegment(url string) string {
	if i := strings.LastIndexByte(url, '/'); i >= 0 {
		return url[i+1:]
	}
	return url
}