controller.go

package main

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"net/http"
	"time"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	clientset "k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/utils/ptr"

	custom "angrybytes.com/external-mysql-operator/pkg/apis/extmysql"
	customv1 "angrybytes.com/external-mysql-operator/pkg/apis/extmysql/v1"
	customclientset "angrybytes.com/external-mysql-operator/pkg/generated/clientset/versioned"
	custominformer "angrybytes.com/external-mysql-operator/pkg/generated/informers/externalversions"
)

const (
	// cacheAge is how long we cache MySQL listings.
	cacheAge = 1 * time.Second
	// actionTimeout is a general timeout used by actions.
	actionTimeout = 20 * time.Second

	managedByLabelKey      = "app.kubernetes.io/managed-by"
	managedByLabelValue    = custom.GroupName
	managedByLabelSelector = managedByLabelKey + "=" + managedByLabelValue
)

var mySQLDBKind = schema.GroupVersionResource{
	Group:    custom.GroupName,
	Version:  "v1",
	Resource: "MySQLDatabase",
}

type Controller struct {
	kube   clientset.Interface
	custom customclientset.Interface
	db     *sql.DB

	informerFactory custominformer.SharedInformerFactory
	dbInformer      cache.SharedIndexInformer
	queue           workqueue.TypedInterface[action]

	dbNames   map[string]struct{}
	userNames map[string]struct{}
	cachedAt  time.Time
}

// action is the interface implemented by queue items
type action interface {
	Handler(c *Controller)
}

func NewController(
	cfg *rest.Config,
	httpClient *http.Client,
	db *sql.DB,
) *Controller {
	kube, err := clientset.NewForConfigAndClient(cfg, httpClient)
	if err != nil {
		log.Fatalf("Erroring building Kubernetes client: %v", err)
	}

	custom, err := customclientset.NewForConfigAndClient(cfg, httpClient)
	if err != nil {
		log.Fatalf("Erroring building Kubernetes client: %v", err)
	}

	informerFactory := custominformer.NewSharedInformerFactory(custom, time.Hour)
	dbInformer := informerFactory.Extmysql().V1().MySQLDatabases().Informer()

	c := &Controller{
		kube:   kube,
		custom: custom,
		db:     db,

		informerFactory: informerFactory,
		dbInformer:      dbInformer,
		queue:           workqueue.NewTyped[action](),

		dbNames:   make(map[string]struct{}),
		userNames: make(map[string]struct{}),
	}

	dbInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj any) {
			c.queue.Add(&updateAction{obj: obj.(*customv1.MySQLDatabase)})
		},
		UpdateFunc: func(old, new any) {
			c.queue.Add(&updateAction{obj: new.(*customv1.MySQLDatabase)})
		},
		DeleteFunc: func(obj any) {
			c.queue.Add(&deleteAction{obj: obj.(*customv1.MySQLDatabase)})
		},
	})

	return c
}

func (c *Controller) Run(ctx context.Context) {
	// Hard fail early if we can't query MySQL.
	if err := c.cacheMySQLState(ctx); err != nil {
		log.Fatalf("Failed to sync MySQL state: %v", err)
	}

	go func() {
		for {
			action, shutdown := c.queue.Get()
			if shutdown {
				break
			}

			action.Handler(c)
			c.queue.Done(action)
		}
	}()

	log.Printf("Controller syncing")
	c.informerFactory.Start(ctx.Done())
	for kind, res := range c.informerFactory.WaitForCacheSync(ctx.Done()) {
		if !res {
			log.Fatalf("Failed to sync %s resources", kind.String())
		}
	}
	syncDone := make(chan struct{})
	c.queue.Add(&syncAction{
		done: syncDone,
	})
	<-syncDone

	log.Printf("Controller started")
	<-ctx.Done()
	log.Printf("Controller stopping")
	c.queue.ShutDown()
}

// cacheMySQLState ensures MySQL listings are fresh.
func (c *Controller) cacheMySQLState(ctx context.Context) error {
	if time.Now().Sub(c.cachedAt) < cacheAge {
		return nil // still fresh
	}

	// Update database listing
	rows, err := c.db.QueryContext(ctx, "SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME LIKE 'k8s_%'")
	if err != nil {
		return fmt.Errorf("Error querying databases: %v", err)
	}

	dbNames := make(map[string]struct{})
	for rows.Next() {
		var dbName string
		if err := rows.Scan(&dbName); err != nil {
			return fmt.Errorf("Error querying databases: %v", err)
		}
		dbNames[dbName] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("Error querying databases: %v", err)
	}

	// Update user listing
	rows, err = c.db.QueryContext(ctx, "SELECT User FROM mysql.user WHERE Host = '%' AND User LIKE 'k8s_%'")
	if err != nil {
		return fmt.Errorf("Error querying users: %v", err)
	}

	userNames := make(map[string]struct{})
	for rows.Next() {
		var userName string
		if err := rows.Scan(&userName); err != nil {
			return fmt.Errorf("Error querying users: %v", err)
		}
		userNames[userName] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("Error querying users: %v", err)
	}

	c.dbNames = dbNames
	c.userNames = userNames
	c.cachedAt = time.Now()

	return nil
}

// setReady updates the Ready condition on a MySQLDatabase
func (c *Controller) setReady(
	ctx context.Context,
	obj *customv1.MySQLDatabase,
	ready bool,
	message string,
) {
	obj = obj.DeepCopy()

	condition := metav1.Condition{
		Type:               "Ready",
		Status:             metav1.ConditionFalse,
		Message:            message,
		ObservedGeneration: obj.Generation,
		LastTransitionTime: metav1.Now(),
	}
	if ready {
		condition.Status = metav1.ConditionTrue
	}

	ok := false
	for idx, entry := range obj.Status.Conditions {
		if entry.Type == "Ready" {
			obj.Status.Conditions[idx] = condition
			ok = true
			break
		}
	}
	if !ok {
		obj.Status.Conditions = append(obj.Status.Conditions, condition)
	}

	dbsApi := c.custom.ExtmysqlV1().MySQLDatabases(obj.Namespace)
	_, err := dbsApi.UpdateStatus(ctx, obj, metav1.UpdateOptions{FieldManager: custom.GroupName})
	if err != nil {
		name := formatResourceName(obj.Namespace, obj.Name)
		log.Printf("Error updating status of MySQLDatabase %s: %v", name, err)
	}
}

// updateAction handles add and update events for MySQLDatabase
type updateAction struct {
	obj *customv1.MySQLDatabase
}

func (act *updateAction) Handler(c *Controller) {
	ctx, cancel := context.WithTimeout(context.Background(), actionTimeout)
	defer cancel()

	obj := act.obj

	name := formatMySQLName(obj)
	_, hasUser := c.userNames[name]
	_, hasDb := c.dbNames[name]
	if hasUser && hasDb {
		return
	}

	if !hasDb {
		_, err := c.db.ExecContext(ctx, fmt.Sprintf("CREATE DATABASE `%s`", name))
		if err != nil {
			log.Printf("Error creating MySQL database %s: %v", name, err)
			c.setReady(ctx, obj, false, "Error creating MySQL database")
			return
		}

		// Update cache
		log.Printf("Created MySQL database %s", name)
		c.dbNames[name] = struct{}{}
	}

	if !hasUser {
		row := c.db.QueryRowContext(ctx, fmt.Sprintf("CREATE USER '%s'@'%%' IDENTIFIED BY RANDOM PASSWORD", name))
		var password, dummy string
		if err := row.Scan(&dummy, &dummy, &password, &dummy); err != nil {
			log.Printf("Error creating MySQL user %s: %v", name, err)
			c.setReady(ctx, obj, false, "Error creating MySQL user")
			return
		}

		_, err := c.db.ExecContext(ctx, fmt.Sprintf("GRANT ALL ON `%s`.* TO '%s'@'%%'", name, name))
		if err != nil {
			log.Printf("Error creating MySQL permissions for user %s: %v", name, err)
			c.setReady(ctx, obj, false, "Error creating MySQL permissions")

			// Try revert.
			_, err = c.db.ExecContext(ctx, fmt.Sprintf("DROP USER '%s'@'%%'", name))
			if err != nil {
				log.Printf("Post-failure cleanup of user %s failed: %v", name, err)
				// Don't try again until the admin fixes this manually.
				c.userNames[name] = struct{}{}
			}

			return
		}

		secretName := formatResourceName(obj.Namespace, obj.Spec.SecretName)
		_, err = c.kube.CoreV1().Secrets(obj.Namespace).Create(ctx, &corev1.Secret{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: obj.Namespace,
				Name:      obj.Spec.SecretName,
				OwnerReferences: []metav1.OwnerReference{{
					APIVersion: mySQLDBKind.GroupVersion().String(),
					Kind:       mySQLDBKind.Resource,
					Name:       obj.Name,
					UID:        obj.UID,
					Controller: ptr.To(true),
				}},
				Labels: map[string]string{
					"app.kubernetes.io/managed-by": custom.GroupName,
				},
			},
			Type:      corev1.SecretTypeBasicAuth,
			Immutable: ptr.To(true),
			Data: map[string][]byte{
				corev1.BasicAuthUsernameKey: []byte(name),
				corev1.BasicAuthPasswordKey: []byte(password),
			},
		}, metav1.CreateOptions{FieldManager: custom.GroupName})
		if err != nil {
			log.Printf("Error creating Secret %s: %v", secretName, err)
			c.setReady(ctx, obj, false, "Error creating Secret")

			// Try revert.
			_, err = c.db.ExecContext(ctx, fmt.Sprintf("DROP USER '%s'@'%%'", name))
			if err != nil {
				log.Printf("Post-failure cleanup of user %s failed: %v", name, err)
				// Don't try again until the admin fixes this manually.
				c.userNames[name] = struct{}{}
			}

			return
		}

		// Update cache
		log.Printf("Created MySQL user %s and secret %s", name, secretName)
		c.userNames[name] = struct{}{}
	}

	c.setReady(ctx, obj, true, "Database, user and secret created")
}

// deleteAction handles delete events on MySQLDatabase
type deleteAction struct {
	obj *customv1.MySQLDatabase
}

func (act *deleteAction) Handler(c *Controller) {
	ctx, cancel := context.WithTimeout(context.Background(), actionTimeout)
	defer cancel()

	obj := act.obj
	name := formatMySQLName(obj)

	if _, ok := c.userNames[name]; ok {
		_, err := c.db.ExecContext(ctx, fmt.Sprintf("DROP USER '%s'@'%%'", name))
		if err != nil {
			log.Printf("Error deleting MySQL user %s: %v", name, err)
		} else {
			log.Printf("Deleted MySQL user %s", name)
			delete(c.userNames, name)
		}
	}

	if _, ok := c.dbNames[name]; ok {
		// TODO: Option to archive instead of delete.
		_, err := c.db.ExecContext(ctx, fmt.Sprintf("DROP DATABASE `%s`", name))
		if err != nil {
			log.Printf("Error deleting MySQL database %s: %v", name, err)
		} else {
			log.Printf("Deleted MySQL database %s", name)
			delete(c.dbNames, name)
		}
	}

	// Delete the secret if it still exists and still references us as owner.
	secretsApi := c.kube.CoreV1().Secrets(obj.Namespace)
	secret, err := secretsApi.Get(ctx, obj.Spec.SecretName, metav1.GetOptions{})
	if err != nil {
		if !errors.IsNotFound(err) {
			log.Printf("Error fetching Secret: %v", err)
		}
	} else if len(secret.OwnerReferences) == 1 && secret.OwnerReferences[0].UID == obj.UID {
		secretName := formatResourceName(secret.Namespace, secret.Name)
		err := secretsApi.Delete(ctx, secret.Name, metav1.DeleteOptions{
			Preconditions: &metav1.Preconditions{
				UID: &secret.UID,
			},
		})
		if err != nil {
			log.Printf("Error deleting Secret %s: %v", secretName, err)
		} else {
			log.Printf("Deleted Secret %s", secretName)
		}
	}
}

// syncAction runs once after informers have synced.
type syncAction struct {
	done chan struct{}
}

func (act *syncAction) Handler(c *Controller) {
	ctx, cancel := context.WithTimeout(context.Background(), actionTimeout)
	defer cancel()

	// Collect expected databases and secrets.
	dbNames := make(map[string]struct{})
	secretNames := make(map[string]struct{})
	for _, obj := range c.dbInformer.GetStore().List() {
		obj := obj.(*customv1.MySQLDatabase)
		dbName := formatMySQLName(obj)
		dbNames[dbName] = struct{}{}
		secretName := formatResourceName(obj.Namespace, obj.Spec.SecretName)
		secretNames[secretName] = struct{}{}
	}

	// Delete old users.
	for name := range c.userNames {
		if _, ok := dbNames[name]; !ok {
			_, err := c.db.ExecContext(ctx, fmt.Sprintf("DROP USER '%s'@'%%'", name))
			if err != nil {
				log.Printf("Error deleting old MySQL user %s: %v", name, err)
			} else {
				log.Printf("Deleted old MySQL user %s", name)
				delete(c.userNames, name)
			}
		}
	}

	// Delete old databases.
	for name := range c.dbNames {
		if _, ok := dbNames[name]; !ok {
			// TODO: Option to archive instead of delete.
			_, err := c.db.ExecContext(ctx, fmt.Sprintf("DROP DATABASE `%s`", name))
			if err != nil {
				log.Printf("Error deleting old MySQL database %s: %v", name, err)
			} else {
				log.Printf("Deleted old MySQL database %s", name)
				delete(c.dbNames, name)
			}
		}
	}

	// Delete old secrets.
	secrets, err := c.kube.CoreV1().Secrets("").List(ctx, metav1.ListOptions{
		LabelSelector: managedByLabelSelector,
	})
	if err != nil {
		log.Printf("Failed to list existing Secrets: %v", err)
	}
	for _, secret := range secrets.Items {
		secretName := formatResourceName(secret.Namespace, secret.Name)
		if _, ok := secretNames[secretName]; !ok {
			err := c.kube.CoreV1().Secrets(secret.Namespace).Delete(ctx, secret.Name, metav1.DeleteOptions{
				Preconditions: &metav1.Preconditions{
					UID: &secret.UID,
				},
			})
			if err != nil {
				log.Printf("Error deleting old Secret %s: %v", secretName, err)
			} else {
				log.Printf("Deleted old Secret %s", secretName)
			}
		}
	}

	act.done <- struct{}{}
}

// formatMySQLName builds a MySQL user/database name from a MySQLDatabase resource.
//
// Because of restrictions on naming of Kubernetes resources, these names are safe to use in
// queries with only basic quoting.
func formatMySQLName(meta *customv1.MySQLDatabase) string {
	return fmt.Sprintf("k8s_%s_%s", meta.Namespace, meta.Name)
}

// formatResourceName formats a full name for a Kubernetes resource.
func formatResourceName(namespace string, name string) string {
	return fmt.Sprintf("%s/%s", namespace, name)
}