243 lines
8.6 KiB
Go
243 lines
8.6 KiB
Go
package controllers
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"database/sql"
|
|
"fmt"
|
|
"math/big"
|
|
"strings"
|
|
"time"
|
|
|
|
platformv1alpha1 "github.com/example/mysql-database-operator/api/v1alpha1"
|
|
_ "github.com/go-sql-driver/mysql"
|
|
corev1 "k8s.io/api/core/v1"
|
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
|
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
|
|
"sigs.k8s.io/controller-runtime/pkg/log"
|
|
"sigs.k8s.io/controller-runtime/pkg/reconcile"
|
|
)
|
|
|
|
const finalizer = "platform.example.com/mysqldatabase-finalizer"
|
|
|
|
type MySQLDatabaseReconciler struct{ client.Client }
|
|
|
|
func (r *MySQLDatabaseReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
|
|
logger := log.FromContext(ctx)
|
|
var obj platformv1alpha1.MySQLDatabase
|
|
if err := r.Get(ctx, req.NamespacedName, &obj); err != nil {
|
|
if apierrors.IsNotFound(err) {
|
|
return reconcile.Result{}, nil
|
|
}
|
|
return reconcile.Result{}, err
|
|
}
|
|
if obj.Spec.DatabaseName == "" {
|
|
return r.setError(ctx, &obj, "InvalidSpec", "databaseName is required")
|
|
}
|
|
if obj.Spec.AdminSecretRef.Name == "" {
|
|
return r.setError(ctx, &obj, "InvalidSpec", "adminSecretRef.name is required")
|
|
}
|
|
|
|
if !obj.DeletionTimestamp.IsZero() {
|
|
if !controllerutil.ContainsFinalizer(&obj, finalizer) {
|
|
return reconcile.Result{}, nil
|
|
}
|
|
if strings.EqualFold(obj.Spec.DeletionPolicy, "Delete") {
|
|
if err := r.withDB(ctx, &obj, func(db *sql.DB) error {
|
|
_, err := db.ExecContext(ctx, "DROP DATABASE IF EXISTS `"+quoteIdent(obj.Spec.DatabaseName)+"`")
|
|
return err
|
|
}); err != nil {
|
|
return reconcile.Result{RequeueAfter: 10 * time.Second}, err
|
|
}
|
|
}
|
|
controllerutil.RemoveFinalizer(&obj, finalizer)
|
|
return reconcile.Result{}, r.Update(ctx, &obj)
|
|
}
|
|
if !controllerutil.ContainsFinalizer(&obj, finalizer) {
|
|
controllerutil.AddFinalizer(&obj, finalizer)
|
|
return reconcile.Result{}, r.Update(ctx, &obj)
|
|
}
|
|
if err := r.withDB(ctx, &obj, func(db *sql.DB) error { return ensureDatabase(ctx, db, &obj) }); err != nil {
|
|
return r.setError(ctx, &obj, "ConnectionError", err.Error())
|
|
}
|
|
if obj.Spec.User != nil && obj.Spec.User.Create {
|
|
username := obj.Spec.User.Name
|
|
if username == "" {
|
|
return r.setError(ctx, &obj, "InvalidSpec", "user.name is required when user.create is true")
|
|
}
|
|
password, err := r.ensureCredentials(ctx, &obj, username)
|
|
if err != nil {
|
|
return r.setError(ctx, &obj, "CredentialsError", err.Error())
|
|
}
|
|
if err := r.withDB(ctx, &obj, func(db *sql.DB) error { return ensureUserAndGrants(ctx, db, &obj, password) }); err != nil {
|
|
return r.setError(ctx, &obj, "UserError", err.Error())
|
|
}
|
|
}
|
|
obj.Status.Phase = "Ready"
|
|
obj.Status.Message = "Database reconciled"
|
|
obj.Status.DatabaseName = obj.Spec.DatabaseName
|
|
if obj.Spec.User != nil {
|
|
obj.Status.UserName = obj.Spec.User.Name
|
|
}
|
|
if obj.Spec.Credentials != nil {
|
|
obj.Status.SecretName = obj.Spec.Credentials.SecretName
|
|
}
|
|
obj.Status.ObservedGeneration = obj.Generation
|
|
obj.Status.Conditions = []metav1.Condition{{Type: "Ready", Status: metav1.ConditionTrue, Reason: "Reconciled", Message: "MySQL database is ready", LastTransitionTime: metav1.Now()}}
|
|
if err := r.Status().Update(ctx, &obj); err != nil {
|
|
logger.Error(err, "status update failed")
|
|
return reconcile.Result{}, err
|
|
}
|
|
return reconcile.Result{RequeueAfter: 5 * time.Minute}, nil
|
|
}
|
|
|
|
func (r *MySQLDatabaseReconciler) setError(ctx context.Context, obj *platformv1alpha1.MySQLDatabase, reason, msg string) (reconcile.Result, error) {
|
|
obj.Status.Phase = "Error"
|
|
obj.Status.Message = msg
|
|
obj.Status.ObservedGeneration = obj.Generation
|
|
obj.Status.Conditions = []metav1.Condition{{Type: "Ready", Status: metav1.ConditionFalse, Reason: reason, Message: msg, LastTransitionTime: metav1.Now()}}
|
|
_ = r.Status().Update(ctx, obj)
|
|
return reconcile.Result{RequeueAfter: 15 * time.Second}, fmt.Errorf("%s: %s", reason, msg)
|
|
}
|
|
|
|
func (r *MySQLDatabaseReconciler) withDB(ctx context.Context, obj *platformv1alpha1.MySQLDatabase, fn func(*sql.DB) error) error {
|
|
var admin corev1.Secret
|
|
if err := r.Get(ctx, types.NamespacedName{Name: obj.Spec.AdminSecretRef.Name, Namespace: obj.Namespace}, &admin); err != nil {
|
|
return err
|
|
}
|
|
username, password := string(admin.Data["username"]), string(admin.Data["password"])
|
|
if username == "" || password == "" {
|
|
return fmt.Errorf("admin secret %q must contain username and password", obj.Spec.AdminSecretRef.Name)
|
|
}
|
|
host := obj.Spec.Connection.Host
|
|
if host == "" {
|
|
host = obj.Spec.ClusterRef.Name
|
|
}
|
|
port := obj.Spec.Connection.Port
|
|
if port == 0 {
|
|
port = 3306
|
|
}
|
|
dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/?parseTime=true&timeout=5s&readTimeout=10s&writeTimeout=10s", username, password, host, port)
|
|
db, err := sql.Open("mysql", dsn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer db.Close()
|
|
pingCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
|
|
defer cancel()
|
|
if err := db.PingContext(pingCtx); err != nil {
|
|
return err
|
|
}
|
|
return fn(db)
|
|
}
|
|
|
|
func ensureDatabase(ctx context.Context, db *sql.DB, obj *platformv1alpha1.MySQLDatabase) error {
|
|
q := "CREATE DATABASE IF NOT EXISTS `" + quoteIdent(obj.Spec.DatabaseName) + "`"
|
|
if obj.Spec.Charset != "" {
|
|
q += " CHARACTER SET " + safeSQLWord(obj.Spec.Charset)
|
|
}
|
|
if obj.Spec.Collation != "" {
|
|
q += " COLLATE " + safeSQLWord(obj.Spec.Collation)
|
|
}
|
|
_, err := db.ExecContext(ctx, q)
|
|
return err
|
|
}
|
|
|
|
func ensureUserAndGrants(ctx context.Context, db *sql.DB, obj *platformv1alpha1.MySQLDatabase, password string) error {
|
|
u := quoteIdent(obj.Spec.User.Name)
|
|
// User is intentionally scoped to localhost wildcard host. For networked K8s apps, '%' is needed.
|
|
if _, err := db.ExecContext(ctx, "CREATE USER IF NOT EXISTS '"+quoteString(obj.Spec.User.Name)+"'@'%' IDENTIFIED BY '"+quoteString(password)+"'"); err != nil {
|
|
return err
|
|
}
|
|
if _, err := db.ExecContext(ctx, "ALTER USER '"+quoteString(obj.Spec.User.Name)+"'@'%' IDENTIFIED BY '"+quoteString(password)+"'"); err != nil {
|
|
return err
|
|
}
|
|
grants := obj.Spec.Privileges
|
|
if len(grants) == 0 {
|
|
grants = []string{"SELECT", "INSERT", "UPDATE", "DELETE"}
|
|
}
|
|
for _, g := range grants {
|
|
if !isAllowedPrivilege(g) {
|
|
return fmt.Errorf("unsupported privilege %q", g)
|
|
}
|
|
}
|
|
grant := strings.Join(grants, ", ")
|
|
_, err := db.ExecContext(ctx, "GRANT "+grant+" ON `"+quoteIdent(obj.Spec.DatabaseName)+"`.* TO '"+quoteString(obj.Spec.User.Name)+"'@'%'")
|
|
_ = u
|
|
return err
|
|
}
|
|
|
|
func (r *MySQLDatabaseReconciler) ensureCredentials(ctx context.Context, obj *platformv1alpha1.MySQLDatabase, username string) (string, error) {
|
|
if obj.Spec.Credentials == nil || obj.Spec.Credentials.SecretName == "" {
|
|
return "", fmt.Errorf("credentials.secretName is required when user.create is true")
|
|
}
|
|
key := types.NamespacedName{Name: obj.Spec.Credentials.SecretName, Namespace: obj.Namespace}
|
|
var s corev1.Secret
|
|
err := r.Get(ctx, key, &s)
|
|
if err == nil {
|
|
p := string(s.Data["password"])
|
|
if p == "" {
|
|
return "", fmt.Errorf("credentials secret must contain password")
|
|
}
|
|
if string(s.Data["username"]) == "" {
|
|
s.Data["username"] = []byte(username)
|
|
if err := r.Update(ctx, &s); err != nil {
|
|
return "", err
|
|
}
|
|
}
|
|
return p, nil
|
|
}
|
|
if !apierrors.IsNotFound(err) {
|
|
return "", err
|
|
}
|
|
p, err := randomPassword(32)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
s = corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: key.Namespace}, Type: corev1.SecretTypeOpaque, StringData: map[string]string{"username": username, "password": p, "database": obj.Spec.DatabaseName}}
|
|
if err := controllerutil.SetControllerReference(obj, &s, r.Scheme); err != nil {
|
|
return "", err
|
|
}
|
|
if err := r.Create(ctx, &s); err != nil {
|
|
return "", err
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
func randomPassword(n int) (string, error) {
|
|
const chars = "ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz23456789!@#$%^&*_-"
|
|
b := make([]byte, n)
|
|
for i := range b {
|
|
x, e := rand.Int(rand.Reader, big.NewInt(int64(len(chars))))
|
|
if e != nil {
|
|
return "", e
|
|
}
|
|
b[i] = chars[x.Int64()]
|
|
}
|
|
return string(b), nil
|
|
}
|
|
func quoteIdent(s string) string { return strings.ReplaceAll(s, "`", "``") }
|
|
func quoteString(s string) string {
|
|
return strings.ReplaceAll(strings.ReplaceAll(s, "\\", "\\\\"), "'", "\\'")
|
|
}
|
|
func safeSQLWord(s string) string {
|
|
for _, r := range s {
|
|
if !(r == '_' || r == '-' || r == '.' || r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9') {
|
|
return "utf8mb4"
|
|
}
|
|
}
|
|
return s
|
|
}
|
|
func isAllowedPrivilege(s string) bool {
|
|
switch strings.ToUpper(strings.TrimSpace(s)) {
|
|
case "SELECT", "INSERT", "UPDATE", "DELETE", "CREATE", "ALTER", "INDEX", "DROP", "REFERENCES", "EXECUTE", "SHOW VIEW", "TRIGGER":
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|