diff --git a/hack/scripts/verify-custom-ca-e2e.sh b/hack/scripts/verify-custom-ca-e2e.sh index 66819016..5f90e072 100755 --- a/hack/scripts/verify-custom-ca-e2e.sh +++ b/hack/scripts/verify-custom-ca-e2e.sh @@ -5,6 +5,7 @@ set -euo pipefail NAMESPACE="wandb-ca-e2e" NAME="wandb" API_APP="api" +MYSQL_INSTANCE="default" TIMEOUT="20m" POLL_SECONDS=10 @@ -79,6 +80,18 @@ if actual != expected: PY } +url_param() { + local url="$1" + local key="$2" + python3 - "${url}" "${key}" <<'PY' +import sys +from urllib.parse import parse_qs, urlparse + +url, key = sys.argv[1:] +print(parse_qs(urlparse(url).query).get(key, [""])[0]) +PY +} + json_has_env() { local name="$1" jq -e --arg name "${name}" '[.spec.containers[]?.env[]? | select(.name == $name)] | length > 0' >/dev/null @@ -203,7 +216,7 @@ wait_until "WeightsAndBiases ${NAMESPACE}/${NAME} status.ready=true" check_wandb wandb_json="$(kubectl -n "${NAMESPACE}" get weightsandbiases.apps.wandb.com "${NAME}" -o json)" inline_ca_count="$(echo "${wandb_json}" | jq -r '(.spec.global.customCACerts // []) | length')" USER_CONFIGMAP="$(echo "${wandb_json}" | jq -r '.spec.global.caCertsConfigMap // ""')" -mysql_ca_enabled="$(echo "${wandb_json}" | jq -r '(((.spec.mysql.externalMysql.sslCa.name // "") | length) > 0 and ((.spec.mysql.externalMysql.sslCa.key // "") | length) > 0)')" +mysql_ca_enabled="$(echo "${wandb_json}" | jq -r --arg instance "${MYSQL_INSTANCE}" '(((.spec.mysql[$instance].externalMysql.sslCa.name // "") | length) > 0 and ((.spec.mysql[$instance].externalMysql.sslCa.key // "") | length) > 0)')" redis_ca_enabled="$(echo "${wandb_json}" | jq -r '(((.spec.redis.externalRedis.sslCa.name // "") | length) > 0 and ((.spec.redis.externalRedis.sslCa.key // "") | length) > 0)')" if [[ "${inline_ca_count}" == "0" && -z "${USER_CONFIGMAP}" ]]; then @@ -219,9 +232,16 @@ if [[ -n "${USER_CONFIGMAP}" ]]; then fi if [[ "${mysql_ca_enabled}" == "true" ]]; then - mysql_url="$(secret_value wandb-mysql-connection url)" + # The connection bundle Secret is named per instance; take it from status + # rather than reconstructing the instance fingerprint here. + mysql_secret="$(echo "${wandb_json}" | jq -r --arg instance "${MYSQL_INSTANCE}" '.status.mysqlStatus[$instance].connection.url.name // ""')" + [[ -n "${mysql_secret}" ]] || fail "no MySQL connection Secret in status for instance ${MYSQL_INSTANCE}" + mysql_url="$(secret_value "${mysql_secret}" url)" assert_url_param "${mysql_url}" "tls" "custom" - assert_url_param "${mysql_url}" "ssl-ca" "/etc/ssl/certs/mysql_ca.pem" + mysql_ca_path="$(url_param "${mysql_url}" "ssl-ca")" + [[ "${mysql_ca_path}" == /*/ca.pem ]] || fail "unexpected ssl-ca path ${mysql_ca_path} in MySQL URL" + mysql_ca_dir="$(dirname "${mysql_ca_path}")" + mysql_ca_volume="mysql-$(basename "${mysql_ca_dir}")" log "MySQL connection URL includes expected CA parameters" fi @@ -238,9 +258,6 @@ workload_ref="${WORKLOAD_KIND} ${WORKLOAD_NAME}" for env_name in SSL_CERT_FILE SSL_CERT_DIR REQUESTS_CA_BUNDLE; do echo "${WORKLOAD_TEMPLATE_JSON}" | json_has_env "${env_name}" || fail "missing env ${env_name} on ${workload_ref}" done -if [[ "${mysql_ca_enabled}" == "true" ]]; then - echo "${WORKLOAD_TEMPLATE_JSON}" | json_has_env MYSQL_CA_CERT_PATH || fail "missing env MYSQL_CA_CERT_PATH on ${workload_ref}" -fi echo "${WORKLOAD_TEMPLATE_JSON}" | json_has_volume wandb-ca-certs-root || fail "missing volume wandb-ca-certs-root on ${workload_ref}" echo "${WORKLOAD_TEMPLATE_JSON}" | json_has_mount wandb-ca-certs-root /usr/local/share/ca-certificates/ || @@ -257,8 +274,9 @@ if [[ -n "${USER_CONFIGMAP}" ]]; then fail "missing user CA ConfigMap mount" fi if [[ "${mysql_ca_enabled}" == "true" ]]; then - echo "${WORKLOAD_TEMPLATE_JSON}" | json_has_volume mysql-ca || fail "missing volume mysql-ca on ${workload_ref}" - echo "${WORKLOAD_TEMPLATE_JSON}" | json_has_mount mysql-ca /etc/ssl/certs/mysql_ca.pem || + echo "${WORKLOAD_TEMPLATE_JSON}" | json_has_volume "${mysql_ca_volume}" || + fail "missing volume ${mysql_ca_volume} on ${workload_ref}" + echo "${WORKLOAD_TEMPLATE_JSON}" | json_has_mount "${mysql_ca_volume}" "${mysql_ca_dir}" || fail "missing MySQL CA mount" fi if [[ "${redis_ca_enabled}" == "true" ]]; then @@ -281,7 +299,7 @@ if [[ -n "${workload_pod}" ]]; then pod_checks+=("test -d /usr/local/share/ca-certificates/configmap") fi if [[ "${mysql_ca_enabled}" == "true" ]]; then - pod_checks+=("test -s /etc/ssl/certs/mysql_ca.pem") + pod_checks+=("test -s ${mysql_ca_path}") fi if [[ "${redis_ca_enabled}" == "true" ]]; then pod_checks+=("test -s /etc/ssl/certs/redis_ca.pem") diff --git a/internal/controller/infra/external/mysql/mysql.go b/internal/controller/infra/external/mysql/mysql.go index 8ed76000..87577e12 100644 --- a/internal/controller/infra/external/mysql/mysql.go +++ b/internal/controller/infra/external/mysql/mysql.go @@ -2,34 +2,16 @@ package mysql import ( "context" - "fmt" - "net/url" apiv2 "github.com/wandb/operator/api/v2" "github.com/wandb/operator/internal/controller/infra/external" + "github.com/wandb/operator/internal/controller/infra/mysqlconnection" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/types" - "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" ) -const ConnectionSecretName = "wandb-mysql-connection" -const caCertPath = "/etc/ssl/certs/mysql_ca.pem" -const sslCertPath = "/etc/ssl/certs/mysql_ssl_cert.pem" -const sslKeyPath = "/etc/ssl/certs/mysql_ssl_key.pem" - -// connectionSecretName returns the connection secret name for an instance. The -// reserved default instance keeps the historical name for backward -// compatibility; other instances are suffixed with their key. -func connectionSecretName(key string) string { - if key == "" || key == apiv2.DefaultInstanceName { - return ConnectionSecretName - } - return fmt.Sprintf("%s-%s", ConnectionSecretName, key) -} - func WriteState( ctx context.Context, c client.Client, @@ -54,41 +36,87 @@ func WriteState( data, err := external.ResolveFields(ctx, c, wandb.Namespace, fields) if err != nil { logger.Error(err, "failed to resolve external mysql fields") - return []metav1.Condition{{ - Type: "Reconciled", - Status: metav1.ConditionFalse, - Reason: "ApiError", - }} + return []metav1.Condition{ + { + Type: mysqlconnection.ProviderReadyType, + Status: metav1.ConditionFalse, + Reason: "SourceSecretsUnavailable", + Message: err.Error(), + }, + { + Type: mysqlconnection.ConnectionResolvedType, + Status: metav1.ConditionFalse, + Reason: "SourceSecretsUnavailable", + Message: err.Error(), + }, + { + Type: mysqlconnection.BundleReadyType, + Status: metav1.ConditionFalse, + Reason: "ConnectionNotResolved", + }, + { + Type: "Reconciled", + Status: metav1.ConditionFalse, + Reason: "ApiError", + }, + } } - dbUrl := url.URL{ - Scheme: "mysql", - Host: fmt.Sprintf("%s:%s", data["Host"], data["Port"]), - User: url.UserPassword(data["Username"], data["Password"]), - Path: data["Database"], + material := mysqlconnection.Material{ + Host: data["Host"], + Port: data["Port"], + Database: data["Database"], + Username: data["Username"], + Password: data["Password"], + TLS: data["Tls"], + CACert: []byte(data["SslCa"]), + ClientCert: []byte(data["SslCert"]), + ClientKey: []byte(data["SslKey"]), } - values := dbUrl.Query() - if tls, ok := data["Tls"]; ok { - values.Set("tls", tls) - } - if _, ok := data["SslCa"]; ok { - if values.Get("tls") == "" { - values.Set("tls", "custom") + if _, err := mysqlconnection.Write(ctx, c, wandb, key, material); err != nil { + logger.Error(err, "failed to write external mysql connection bundle") + return []metav1.Condition{ + { + Type: mysqlconnection.ProviderReadyType, + Status: metav1.ConditionTrue, + Reason: "SourceSecretsResolved", + }, + { + Type: mysqlconnection.ConnectionResolvedType, + Status: metav1.ConditionFalse, + Reason: "InvalidConnection", + Message: err.Error(), + }, + { + Type: mysqlconnection.BundleReadyType, + Status: metav1.ConditionFalse, + Reason: "InvalidConnection", + }, + { + Type: "Reconciled", + Status: metav1.ConditionFalse, + Reason: "InvalidConnection", + }, } - values.Set("ssl-ca", caCertPath) - } - if _, ok := data["SslCert"]; ok { - values.Set("ssl-cert", sslCertPath) } - if _, ok := data["SslKey"]; ok { - values.Set("ssl-key", sslKeyPath) - } - dbUrl.RawQuery = values.Encode() - - data["url"] = dbUrl.String() - nsName := types.NamespacedName{Namespace: wandb.Namespace, Name: connectionSecretName(key)} - return external.WriteConnectionSecret(ctx, c, wandb, nsName, data) + return []metav1.Condition{ + { + Type: mysqlconnection.ProviderReadyType, + Status: metav1.ConditionTrue, + Reason: "SourceSecretsResolved", + }, + { + Type: mysqlconnection.ConnectionResolvedType, + Status: metav1.ConditionTrue, + Reason: "SourceSecretsResolved", + }, + { + Type: mysqlconnection.BundleReadyType, + Status: metav1.ConditionTrue, + Reason: "BundleWritten", + }, + } } func ReadState( @@ -98,30 +126,17 @@ func ReadState( key string, newConditions []metav1.Condition, ) ([]metav1.Condition, *apiv2.MysqlConnection) { - nsName := types.NamespacedName{Namespace: wandb.Namespace, Name: connectionSecretName(key)} - _, conditions, found := external.ReadConnectionSecret(ctx, c, nsName, newConditions) - if !found { - return conditions, nil - } - - localRef := corev1.LocalObjectReference{Name: nsName.Name} - return conditions, &apiv2.MysqlConnection{ - URL: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "url", Optional: ptr.To(false)}, - Host: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Host", Optional: ptr.To(false)}, - Port: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Port", Optional: ptr.To(false)}, - Database: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Database", Optional: ptr.To(false)}, - Username: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Username", Optional: ptr.To(false)}, - Password: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Password", Optional: ptr.To(false)}, - Tls: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Tls", Optional: ptr.To(true)}, - SslCa: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "SslCa", Optional: ptr.To(true)}, - SslCert: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "SslCert", Optional: ptr.To(true)}, - SslKey: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "SslKey", Optional: ptr.To(true)}, + connection, err := mysqlconnection.Read(ctx, c, wandb, key) + if err != nil { + return append(newConditions, metav1.Condition{ + Type: "Reconciled", + Status: metav1.ConditionFalse, + Reason: "ApiError", + }), nil } + return newConditions, connection } func DeleteConnectionSecret(ctx context.Context, c client.Client, wandb *apiv2.WeightsAndBiases, key string) error { - return external.DeleteConnectionSecret(ctx, c, types.NamespacedName{ - Namespace: wandb.Namespace, - Name: connectionSecretName(key), - }) + return mysqlconnection.Delete(ctx, c, wandb, key) } diff --git a/internal/controller/infra/external/mysql/mysql_test.go b/internal/controller/infra/external/mysql/mysql_test.go index 77a82965..a4971319 100644 --- a/internal/controller/infra/external/mysql/mysql_test.go +++ b/internal/controller/infra/external/mysql/mysql_test.go @@ -2,11 +2,19 @@ package mysql import ( "context" + "crypto/ed25519" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" "net/url" "testing" + "time" "github.com/stretchr/testify/require" apiv2 "github.com/wandb/operator/api/v2" + "github.com/wandb/operator/internal/controller/infra/mysqlconnection" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -36,9 +44,7 @@ func TestWriteStateAddsCustomTLSParamsWhenCACertPresent(t *testing.T) { "Database": []byte("wandb"), "Username": []byte("wandb"), "Password": []byte("secret"), - "SslCa": []byte("---ca---"), - "SslCert": []byte("---cert---"), - "SslKey": []byte("---key---"), + "SslCa": testCACertificate(t), }, } wandb := &apiv2.WeightsAndBiases{ @@ -53,8 +59,6 @@ func TestWriteStateAddsCustomTLSParamsWhenCACertPresent(t *testing.T) { Username: mysqlSel("Username"), Password: mysqlSel("Password"), SslCa: mysqlSel("SslCa"), - SslCert: mysqlSel("SslCert"), - SslKey: mysqlSel("SslKey"), }, }}, }, @@ -62,10 +66,27 @@ func TestWriteStateAddsCustomTLSParamsWhenCACertPresent(t *testing.T) { client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(wandb, source).Build() conditions := WriteState(context.Background(), client, wandb, apiv2.DefaultInstanceName, wandb.Spec.MySQL[apiv2.DefaultInstanceName].ExternalMysql) - require.Nil(t, conditions) + require.Len(t, conditions, 3) + for _, conditionType := range []string{ + mysqlconnection.ProviderReadyType, + mysqlconnection.ConnectionResolvedType, + mysqlconnection.BundleReadyType, + } { + require.Condition(t, func() bool { + for _, condition := range conditions { + if condition.Type == conditionType { + return condition.Status == metav1.ConditionTrue + } + } + return false + }) + } written := &corev1.Secret{} - require.NoError(t, client.Get(context.Background(), types.NamespacedName{Name: ConnectionSecretName, Namespace: "default"}, written)) + require.NoError(t, client.Get(context.Background(), types.NamespacedName{ + Name: mysqlconnection.SecretName(wandb.Name, apiv2.DefaultInstanceName), + Namespace: "default", + }, written)) data := mysqlConnectionData(written) parsed, err := url.Parse(data["url"]) require.NoError(t, err) @@ -73,9 +94,13 @@ func TestWriteStateAddsCustomTLSParamsWhenCACertPresent(t *testing.T) { require.Equal(t, "mysql.example.com:3306", parsed.Host) require.Equal(t, "/wandb", parsed.Path) require.Equal(t, "custom", parsed.Query().Get("tls")) - require.Equal(t, caCertPath, parsed.Query().Get("ssl-ca")) - require.Equal(t, sslCertPath, parsed.Query().Get("ssl-cert")) - require.Equal(t, sslKeyPath, parsed.Query().Get("ssl-key")) + require.Equal( + t, + mysqlconnection.MountPath(apiv2.DefaultInstanceName)+"/"+mysqlconnection.CACertFile, + parsed.Query().Get("ssl-ca"), + ) + require.Empty(t, parsed.Query().Get("ssl-cert")) + require.Equal(t, mysqlconnection.BundleVersion, written.Annotations[mysqlconnection.BundleVersionAnnotation]) } func mysqlConnectionData(secret *corev1.Secret) map[string]string { @@ -88,3 +113,20 @@ func mysqlConnectionData(secret *corev1.Secret) map[string]string { } return out } + +func testCACertificate(t *testing.T) []byte { + t.Helper() + publicKey, privateKey, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "test CA"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + KeyUsage: x509.KeyUsageCertSign, + } + certificate, err := x509.CreateCertificate(rand.Reader, template, template, publicKey, privateKey) + require.NoError(t, err) + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certificate}) +} diff --git a/internal/controller/infra/managed/mysql/moco/conn.go b/internal/controller/infra/managed/mysql/moco/conn.go deleted file mode 100644 index 918a3b1a..00000000 --- a/internal/controller/infra/managed/mysql/moco/conn.go +++ /dev/null @@ -1,103 +0,0 @@ -package moco - -import ( - "context" - "errors" - "fmt" - "net/url" - - apiv2 "github.com/wandb/operator/api/v2" - "github.com/wandb/operator/internal/controller/common" - corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime/schema" - "k8s.io/utils/ptr" - "sigs.k8s.io/controller-runtime/pkg/client" -) - -type mysqlConnInfo struct { - Host string - Port string - User string - Password string - Database string -} - -func (c *mysqlConnInfo) toURL() string { - password := url.QueryEscape(c.Password) - return fmt.Sprintf("mysql://%s:%s@%s:%s/%s", c.User, password, c.Host, c.Port, c.Database) -} - -func writeMySQLConnInfo( - ctx context.Context, - client client.Client, - owner client.Object, - nsnBuilder *NsNameBuilder, - connInfo *mysqlConnInfo, -) ( - *apiv2.MysqlConnection, error, -) { - var err error - var found bool - var gvk schema.GroupVersionKind - var actual = &corev1.Secret{} - - if connInfo == nil { - return nil, errors.New("missing connection info") - } - - nsName := nsnBuilder.ConnectionNsName() - urlKey := "url" - - if found, err = common.GetResource( - ctx, client, nsName, AppConnTypeName, actual, - ); err != nil { - return nil, err - } - if !found { - actual = nil - } - - if gvk, err = client.GroupVersionKindFor(owner); err != nil { - return nil, fmt.Errorf("could not get GVK for owner: %w", err) - } - ref := metav1.OwnerReference{ - APIVersion: gvk.GroupVersion().String(), - Kind: gvk.Kind, - Name: owner.GetName(), - UID: owner.GetUID(), - Controller: ptr.To(false), - BlockOwnerDeletion: ptr.To(false), - } - - desired := &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: nsName.Name, - Namespace: nsName.Namespace, - OwnerReferences: []metav1.OwnerReference{ref}, - }, - Type: corev1.SecretTypeOpaque, - StringData: map[string]string{ - urlKey: connInfo.toURL(), - "Host": connInfo.Host, - "Port": connInfo.Port, - "Database": connInfo.Database, - "Username": connInfo.User, - "Password": connInfo.Password, - }, - } - - if _, err = common.CrudResource(ctx, client, desired, actual); err != nil { - return nil, err - } - - localRef := corev1.LocalObjectReference{Name: nsName.Name} - return &apiv2.MysqlConnection{ - URL: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: urlKey, Optional: ptr.To(false)}, - Host: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Host", Optional: ptr.To(false)}, - Port: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Port", Optional: ptr.To(false)}, - Database: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Database", Optional: ptr.To(false)}, - Username: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Username", Optional: ptr.To(false)}, - Password: corev1.SecretKeySelector{LocalObjectReference: localRef, Key: "Password", Optional: ptr.To(false)}, - }, nil -} diff --git a/internal/controller/infra/managed/mysql/moco/detach.go b/internal/controller/infra/managed/mysql/moco/detach.go index 03d4c8b3..599d1c2a 100644 --- a/internal/controller/infra/managed/mysql/moco/detach.go +++ b/internal/controller/infra/managed/mysql/moco/detach.go @@ -27,6 +27,19 @@ func CheckDetached( if err != nil || !found { return nil } + if actual.Annotations[DetachedAnnotation] == "true" { + return []metav1.Condition{ + { + Type: common.ReconciledType, + Status: metav1.ConditionFalse, + Reason: common.DetachedSpecMismatch, + Message: "detached MySQL CR must be renamed before it can be managed again", + }, + } + } + if actual.Labels[WandbUIDLabel] == string(wandbUID) { + return nil + } if !common.IsDetached(actual, wandbUID) { return nil } @@ -63,37 +76,98 @@ func DetachFinalizer( return nil } - if common.IsDetached(actual, wandbOwner.GetUID()) { + if markDetached(actual, wandbOwner.GetUID()) { + if err = cl.Update(ctx, actual); err != nil { + if errors.IsNotFound(err) { + return nil + } + log.Error("error detaching MysqlCluster CR", logx.ErrAttr(err)) + return err + } + log.Info("detached MysqlCluster CR", "name", actual.Name) + } else { log.Debug("MysqlCluster CR already detached") - return nil } - common.RemoveOwnerReference(actual, wandbOwner.GetUID()) - if err = cl.Update(ctx, actual); err != nil { - if errors.IsNotFound(err) { - return nil - } - log.Error("error detaching MysqlCluster CR", logx.ErrAttr(err)) + configMap := &corev1.ConfigMap{} + found, err = common.GetResource( + ctx, + cl, + types.NamespacedName{Namespace: specNamespacedName.Namespace, Name: MyCnfConfigMapName(specNamespacedName.Name)}, + "ConfigMap", + configMap, + ) + if err != nil || !found { return err } - log.Info("detached MysqlCluster CR", "name", actual.Name) - - if err = detachConnectionSecret(ctx, cl, nsnBuilder, wandbOwner); err != nil { - log.Error("error detaching connection secret", logx.ErrAttr(err)) - return err + if markDetached(configMap, wandbOwner.GetUID()) { + if err := cl.Update(ctx, configMap); err != nil && !errors.IsNotFound(err) { + return err + } } + return nil } -func detachConnectionSecret(ctx context.Context, cl client.Client, nsnBuilder *NsNameBuilder, wandbOwner client.Object) error { - secret := &corev1.Secret{} - found, err := common.GetResource(ctx, cl, nsnBuilder.ConnectionNsName(), "Secret", secret) +// ClearDetached re-adopts a cluster (and its config map) that a previous +// reconcile detached, so an instance removed and then restored in the spec does +// not stay permanently unmanageable. +func ClearDetached(ctx context.Context, cl client.Client, specNamespacedName types.NamespacedName) error { + ctx, log := logx.WithSlog(ctx, logx.Mysql) + nsnBuilder := createNsNameBuilder(specNamespacedName) + + cluster := &mocov1beta2.MySQLCluster{} + found, err := common.GetResource(ctx, cl, nsnBuilder.ClusterNsName(), ResourceTypeName, cluster) + if err != nil || !found { + return err + } + if cluster.Annotations[DetachedAnnotation] != "true" { + return nil + } + if err := clearDetachedAnnotation(ctx, cl, cluster); err != nil { + return err + } + log.Info("re-adopted detached MysqlCluster CR", "name", cluster.Name) + + configMap := &corev1.ConfigMap{} + found, err = common.GetResource( + ctx, + cl, + types.NamespacedName{Namespace: specNamespacedName.Namespace, Name: MyCnfConfigMapName(specNamespacedName.Name)}, + "ConfigMap", + configMap, + ) if err != nil || !found { return err } - common.RemoveOwnerReference(secret, wandbOwner.GetUID()) - if err = cl.Update(ctx, secret); err != nil && !errors.IsNotFound(err) { + return clearDetachedAnnotation(ctx, cl, configMap) +} + +func clearDetachedAnnotation(ctx context.Context, cl client.Client, obj client.Object) error { + annotations := obj.GetAnnotations() + if annotations[DetachedAnnotation] == "" { + return nil + } + patch := client.MergeFrom(obj.DeepCopyObject().(client.Object)) + delete(annotations, DetachedAnnotation) + obj.SetAnnotations(annotations) + if err := cl.Patch(ctx, obj, patch); err != nil && !errors.IsNotFound(err) { return err } return nil } + +func markDetached(obj client.Object, ownerUID types.UID) bool { + changed := !common.IsDetached(obj, ownerUID) + common.RemoveOwnerReference(obj, ownerUID) + annotations := obj.GetAnnotations() + if annotations == nil { + annotations = map[string]string{} + } + if annotations[DetachedAnnotation] != "true" { + annotations[DetachedAnnotation] = "true" + obj.SetAnnotations(annotations) + changed = true + } + return changed +} diff --git a/internal/controller/infra/managed/mysql/moco/purge.go b/internal/controller/infra/managed/mysql/moco/purge.go index 0bd0ba84..25787241 100644 --- a/internal/controller/infra/managed/mysql/moco/purge.go +++ b/internal/controller/infra/managed/mysql/moco/purge.go @@ -2,11 +2,15 @@ package moco import ( "context" + "fmt" + "strings" + mocov1beta2 "github.com/cybozu-go/moco/api/v1beta2" "github.com/wandb/operator/internal/controller/common" "github.com/wandb/operator/internal/logx" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" @@ -22,9 +26,71 @@ func PurgeFinalizer( if onDeleteRule.Policy != common.Purge { return nil } + cluster := &mocov1beta2.MySQLCluster{} + if err := cl.Get(ctx, specNamespacedName, cluster); err == nil { + if err := cl.Delete(ctx, cluster); err != nil && !errors.IsNotFound(err) { + return err + } + } else if !errors.IsNotFound(err) { + return err + } + + configMap := &corev1.ConfigMap{} + configMapName := types.NamespacedName{ + Namespace: specNamespacedName.Namespace, + Name: MyCnfConfigMapName(specNamespacedName.Name), + } + if err := cl.Get(ctx, configMapName, configMap); err == nil { + if err := cl.Delete(ctx, configMap); err != nil && !errors.IsNotFound(err) { + return err + } + } else if !errors.IsNotFound(err) { + return err + } + credentials := &corev1.Secret{} + credentialsName := types.NamespacedName{ + Namespace: specNamespacedName.Namespace, + Name: "moco-" + specNamespacedName.Name, + } + if err := cl.Get(ctx, credentialsName, credentials); err == nil { + if err := cl.Delete(ctx, credentials); err != nil && !errors.IsNotFound(err) { + return err + } + } else if !errors.IsNotFound(err) { + return err + } + if err := purgeClusterPVCs(ctx, cl, specNamespacedName); err != nil { + return err + } return purgeAssociatedResources(ctx, cl, specNamespacedName.Namespace, onDeleteRule.Selector) } +// purgeClusterPVCs deletes the PVCs Moco created for this cluster, selected by +// Moco's PVC naming scheme rather than by label, so purging works for clusters +// created before the provenance labels existed without a selector wide enough +// to also match a sibling instance's storage. +func purgeClusterPVCs(ctx context.Context, cl client.Client, specNamespacedName types.NamespacedName) error { + log := logx.GetSlog(ctx) + cluster := &mocov1beta2.MySQLCluster{ObjectMeta: metav1.ObjectMeta{Name: specNamespacedName.Name}} + prefix := fmt.Sprintf("%s-%s-", dataVolumeName, cluster.PrefixedName()) + + pvcList := &corev1.PersistentVolumeClaimList{} + if err := cl.List(ctx, pvcList, &client.ListOptions{Namespace: specNamespacedName.Namespace}); err != nil { + return err + } + for i := range pvcList.Items { + pvc := &pvcList.Items[i] + if !strings.HasPrefix(pvc.Name, prefix) { + continue + } + if err := cl.Delete(ctx, pvc); err != nil && !errors.IsNotFound(err) { + return err + } + log.Info("Purged MySQL cluster PVC", "pvc", pvc.Name) + } + return nil +} + func purgeAssociatedResources( ctx context.Context, cl client.Client, diff --git a/internal/controller/infra/managed/mysql/moco/read.go b/internal/controller/infra/managed/mysql/moco/read.go index 30aa87cd..dd4b65af 100644 --- a/internal/controller/infra/managed/mysql/moco/read.go +++ b/internal/controller/infra/managed/mysql/moco/read.go @@ -5,8 +5,8 @@ import ( "fmt" mocov1beta2 "github.com/cybozu-go/moco/api/v1beta2" - apiv2 "github.com/wandb/operator/api/v2" ctrlcommon "github.com/wandb/operator/internal/controller/common" + "github.com/wandb/operator/internal/controller/infra/mysqlconnection" "github.com/wandb/operator/internal/logx" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -14,7 +14,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" ) -func readConnectionDetails(ctx context.Context, c client.Client, actual *mocov1beta2.MySQLCluster, nn types.NamespacedName) *mysqlConnInfo { +func readConnectionDetails(ctx context.Context, c client.Client, actual *mocov1beta2.MySQLCluster, nn types.NamespacedName) *mysqlconnection.Material { log := logx.GetSlog(ctx) cred := &corev1.Secret{} @@ -27,10 +27,10 @@ func readConnectionDetails(ctx context.Context, c client.Client, actual *mocov1b if pw == "" { return nil } - return &mysqlConnInfo{ + return &mysqlconnection.Material{ Host: fmt.Sprintf("moco-%s-primary.%s.svc.cluster.local", actual.Name, actual.Namespace), Port: "3306", - User: "moco-writable", + Username: "moco-writable", Database: "wandb_local", Password: pw, } @@ -40,9 +40,8 @@ func ReadState( ctx context.Context, k8sClient client.Client, specNamespacedName types.NamespacedName, - wandbOwner client.Object, onDeleteRule ctrlcommon.OnDeleteRule, -) ([]metav1.Condition, *apiv2.MysqlConnection) { +) ([]metav1.Condition, *mysqlconnection.Material) { ctx, _ = logx.WithSlog(ctx, logx.Mysql) log := logx.GetSlog(ctx) @@ -90,32 +89,10 @@ func ReadState( } } - var connection *apiv2.MysqlConnection + var connection *mysqlconnection.Material if actual != nil { - connInfo := readConnectionDetails(ctx, k8sClient, actual, specNamespacedName) - - connection, err = writeMySQLConnInfo( - ctx, k8sClient, wandbOwner, nsnBuilder, connInfo, - ) - if err != nil { - if err.Error() == "missing connection info" { - return []metav1.Condition{ - { - Type: MySQLConnectionInfoType, - Status: metav1.ConditionFalse, - Reason: ctrlcommon.NoResourceReason, - }, - }, nil - } - return []metav1.Condition{ - { - Type: MySQLConnectionInfoType, - Status: metav1.ConditionUnknown, - Reason: ctrlcommon.ApiErrorReason, - }, - }, nil - } + connection = readConnectionDetails(ctx, k8sClient, actual, specNamespacedName) if connection == nil { conditions = append(conditions, metav1.Condition{ Type: MySQLConnectionInfoType, diff --git a/internal/controller/infra/managed/mysql/moco/spec.go b/internal/controller/infra/managed/mysql/moco/spec.go index 89c88c48..2fae3717 100644 --- a/internal/controller/infra/managed/mysql/moco/spec.go +++ b/internal/controller/infra/managed/mysql/moco/spec.go @@ -7,11 +7,13 @@ import ( mococonstants "github.com/cybozu-go/moco/pkg/constants" apiv2 "github.com/wandb/operator/api/v2" "github.com/wandb/operator/internal/controller/common" + "github.com/wandb/operator/internal/controller/infra/mysqlconnection" "github.com/wandb/operator/pkg/utils" "github.com/wandb/operator/pkg/wandb/manifest" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" corev1ac "k8s.io/client-go/applyconfigurations/core/v1" "k8s.io/utils/ptr" @@ -19,10 +21,14 @@ import ( ) const ( - MysqlModuleName = "moco" + MysqlModuleName = "moco" + WandbUIDLabel = "weightsandbiases.apps.wandb.com/uid" + InstanceLabel = "weightsandbiases.apps.wandb.com/mysql-instance-id" + RetentionPolicyAnnotation = "weightsandbiases.apps.wandb.com/retention-policy" + DetachedAnnotation = "weightsandbiases.apps.wandb.com/detached" // Moco names the resulting PVCs "--" - // (= "-moco--"); ensurePVCLabels and purge rely on this. + // (= "-moco--"); ensurePVCMetadata and purge rely on this. dataVolumeName = mococonstants.MySQLDataVolumeName // TODO: remove this hardcoded default once all supported manifest versions @@ -30,6 +36,16 @@ const ( defaultMocoMySQLImage = "ghcr.io/cybozu-go/moco/mysql:8.4.8" ) +// manifestMysqlConfig resolves the manifest infra config for an instance, +// falling back to the manifest "default" entry, matching how the init job +// resolves its image. +func manifestMysqlConfig(mfst manifest.Manifest, instance string) manifest.InfraConfig { + if cfg, ok := mfst.Mysql[instance]; ok { + return cfg + } + return mfst.Mysql[apiv2.DefaultInstanceName] +} + func MocoMySQLImage(img manifest.ImageRef, globalImageRegistry string) string { if out := img.GetImage(globalImageRegistry); out != "" { return out @@ -48,6 +64,7 @@ const ( func ToMocoMySQLClusterSpec( ctx context.Context, + instance string, spec apiv2.ManagedMysqlSpec, wandb *apiv2.WeightsAndBiases, scheme *runtime.Scheme, @@ -60,27 +77,35 @@ func ToMocoMySQLClusterSpec( ObjectMeta: metav1.ObjectMeta{ Name: MyCnfConfigMapName(spec.Name), Namespace: spec.Namespace, - Labels: BuildWandbMysqlLabels(wandb), + Labels: BuildWandbMysqlLabels(wandb, instance), + Annotations: map[string]string{ + RetentionPolicyAnnotation: string(wandb.GetRetentionPolicy(spec.ManagedInfraSpec).OnDelete), + }, }, Data: map[string]string{ "sync_binlog": "1", "innodb_flush_log_at_trx_commit": "1", }, } - if err := controllerutil.SetControllerReference(wandb, cm, scheme); err != nil { - return nil, nil, err + if cm.Namespace == wandb.Namespace { + if err := controllerutil.SetControllerReference(wandb, cm, scheme); err != nil { + return nil, nil, err + } } cluster := &mocov1beta2.MySQLCluster{ ObjectMeta: metav1.ObjectMeta{ Name: spec.Name, Namespace: spec.Namespace, - Labels: BuildWandbMysqlLabels(wandb), + Labels: BuildWandbMysqlLabels(wandb, instance), + Annotations: map[string]string{ + RetentionPolicyAnnotation: string(wandb.GetRetentionPolicy(spec.ManagedInfraSpec).OnDelete), + }, }, Spec: mocov1beta2.MySQLClusterSpec{ Replicas: replicas, MySQLConfigMapName: ptr.To(MyCnfConfigMapName(spec.Name)), PodTemplate: mocov1beta2.PodTemplateSpec{ - Spec: buildMocoPodSpec(spec.Config.Resources, mfst.Mysql["default"].Images["mysql"], wandb), + Spec: buildMocoPodSpec(spec.Config.Resources, manifestMysqlConfig(mfst, instance).Images["mysql"], wandb), OverwriteContainers: mocoOverwriteContainers(), }, VolumeClaimTemplates: []mocov1beta2.PersistentVolumeClaim{ @@ -97,13 +122,15 @@ func ToMocoMySQLClusterSpec( cluster.Spec.Collectors = []string{"engine_innodb_status", "info_schema.innodb_metrics"} } - if err := controllerutil.SetControllerReference(wandb, cluster, scheme); err != nil { - return nil, nil, err + if cluster.Namespace == wandb.Namespace { + if err := controllerutil.SetControllerReference(wandb, cluster, scheme); err != nil { + return nil, nil, err + } } return cluster, cm, nil } -func buildMocoPodSpec(resources corev1.ResourceRequirements, img manifest.ImageRef, wandb *apiv2.WeightsAndBiases,) mocov1beta2.PodSpecApplyConfiguration { +func buildMocoPodSpec(resources corev1.ResourceRequirements, img manifest.ImageRef, wandb *apiv2.WeightsAndBiases) mocov1beta2.PodSpecApplyConfiguration { container := corev1ac.Container(). WithName("mysqld"). WithImage(MocoMySQLImage(img, wandb.Spec.Global.ImageRegistry)). @@ -188,10 +215,17 @@ func buildPVCSpec(storageSize string) mocov1beta2.PersistentVolumeClaimSpecApply return mocov1beta2.PersistentVolumeClaimSpecApplyConfiguration(*pvcSpec) } -func BuildWandbMysqlLabels(wandb *apiv2.WeightsAndBiases) map[string]string { - return common.BuildWandbLabels(wandb, MysqlModuleName) +func BuildWandbMysqlLabels(wandb *apiv2.WeightsAndBiases, instance string) map[string]string { + result := common.BuildWandbLabels(wandb, MysqlModuleName) + result[InstanceLabel] = mysqlconnection.InstanceID(instance) + if wandb.UID != "" { + result[WandbUIDLabel] = string(wandb.UID) + } + return result } -func ToMysqlOnDeleteRule(wandb *apiv2.WeightsAndBiases, retentionPolicy apiv2.RetentionPolicy) common.OnDeleteRule { - return common.ToOnDeleteRule(wandb, retentionPolicy, MysqlModuleName) +func ToMysqlOnDeleteRule(wandb *apiv2.WeightsAndBiases, instance string, retentionPolicy apiv2.RetentionPolicy) common.OnDeleteRule { + rule := common.ToOnDeleteRule(wandb, retentionPolicy, MysqlModuleName) + rule.Selector = labels.SelectorFromSet(BuildWandbMysqlLabels(wandb, instance)) + return rule } diff --git a/internal/controller/infra/managed/mysql/moco/spec_test.go b/internal/controller/infra/managed/mysql/moco/spec_test.go index fa1ac7a2..5a395261 100644 --- a/internal/controller/infra/managed/mysql/moco/spec_test.go +++ b/internal/controller/infra/managed/mysql/moco/spec_test.go @@ -8,9 +8,10 @@ import ( . "github.com/onsi/gomega" "github.com/samber/lo" apiv2 "github.com/wandb/operator/api/v2" - "github.com/wandb/operator/pkg/wandb/manifest" "github.com/wandb/operator/internal/controller/common" + "github.com/wandb/operator/internal/controller/infra/mysqlconnection" "github.com/wandb/operator/pkg/utils" + "github.com/wandb/operator/pkg/wandb/manifest" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -28,6 +29,7 @@ var _ = Describe("Moco MySQL specs", func() { It("renders hardened pod and container security settings", func() { cluster, _, err := ToMocoMySQLClusterSpec( context.Background(), + apiv2.DefaultInstanceName, apiv2.ManagedMysqlSpec{ Name: "mysql", Namespace: "wandb", @@ -65,6 +67,7 @@ var _ = Describe("Moco MySQL specs", func() { cluster, _, err := ToMocoMySQLClusterSpec( context.Background(), + apiv2.DefaultInstanceName, apiv2.ManagedMysqlSpec{ Name: "mysql", Namespace: "wandb", @@ -87,6 +90,7 @@ var _ = Describe("Moco MySQL specs", func() { // A non-empty Collectors list is what makes Moco inject the mysqld_exporter sidecar. enabled, _, err := ToMocoMySQLClusterSpec( context.Background(), + apiv2.DefaultInstanceName, apiv2.ManagedMysqlSpec{ Name: "mysql", Namespace: "wandb", Replicas: 3, StorageSize: "10Gi", Telemetry: apiv2.Telemetry{Enabled: true}, @@ -100,6 +104,7 @@ var _ = Describe("Moco MySQL specs", func() { disabled, _, err := ToMocoMySQLClusterSpec( context.Background(), + apiv2.DefaultInstanceName, apiv2.ManagedMysqlSpec{ Name: "mysql", Namespace: "wandb", Replicas: 3, StorageSize: "10Gi", Telemetry: apiv2.Telemetry{Enabled: false}, @@ -112,6 +117,65 @@ var _ = Describe("Moco MySQL specs", func() { Expect(disabled.Spec.Collectors).To(BeEmpty()) }) + It("uses labels instead of invalid owner references across namespaces", func() { + wandb := mocoWandb() + wandb.UID = "wandb-uid" + cluster, configMap, err := ToMocoMySQLClusterSpec( + context.Background(), + "analytics", + apiv2.ManagedMysqlSpec{ + Name: "mysql", + Namespace: "database", + Replicas: 3, + StorageSize: "10Gi", + }, + wandb, + mocoScheme(), + manifest.Manifest{}, + ) + Expect(err).NotTo(HaveOccurred()) + Expect(cluster.OwnerReferences).To(BeEmpty()) + Expect(configMap.OwnerReferences).To(BeEmpty()) + Expect(cluster.Labels).To(HaveKeyWithValue(WandbUIDLabel, "wandb-uid")) + Expect(cluster.Labels).To(HaveKeyWithValue(InstanceLabel, mysqlconnection.InstanceID("analytics"))) + }) + + It("removes legacy W&B owner references from cross-namespace resources", func() { + wandb := mocoWandb() + wandb.UID = "wandb-uid" + desired, desiredConfigMap, err := ToMocoMySQLClusterSpec( + context.Background(), + "analytics", + apiv2.ManagedMysqlSpec{ + Name: "mysql", Namespace: "database", Replicas: 3, StorageSize: "10Gi", + }, + wandb, + mocoScheme(), + manifest.Manifest{}, + ) + Expect(err).NotTo(HaveOccurred()) + legacyRef := metav1.OwnerReference{ + APIVersion: apiv2.GroupVersion.String(), Kind: "WeightsAndBiases", Name: wandb.Name, UID: wandb.UID, + } + actual := desired.DeepCopy() + actual.OwnerReferences = []metav1.OwnerReference{legacyRef} + actualConfigMap := desiredConfigMap.DeepCopy() + actualConfigMap.OwnerReferences = []metav1.OwnerReference{legacyRef} + cl := fake.NewClientBuilder().WithScheme(mocoScheme()).WithObjects(actual, actualConfigMap).Build() + + WriteState( + context.Background(), cl, types.NamespacedName{Name: "mysql", Namespace: "database"}, + desired, desiredConfigMap, BuildWandbMysqlLabels(wandb, "analytics"), + ) + + gotCluster := &mocov1beta2.MySQLCluster{} + Expect(cl.Get(context.Background(), types.NamespacedName{Name: "mysql", Namespace: "database"}, gotCluster)).To(Succeed()) + Expect(gotCluster.OwnerReferences).To(BeEmpty()) + gotConfigMap := &corev1.ConfigMap{} + Expect(cl.Get(context.Background(), types.NamespacedName{Name: desiredConfigMap.Name, Namespace: "database"}, gotConfigMap)).To(Succeed()) + Expect(gotConfigMap.OwnerReferences).To(BeEmpty()) + }) + DescribeTable("refuses to forward a replica count Moco rejects", func(replicas int32) { ctx := context.Background() @@ -120,6 +184,7 @@ var _ = Describe("Moco MySQL specs", func() { desired, cm, err := ToMocoMySQLClusterSpec( ctx, + apiv2.DefaultInstanceName, apiv2.ManagedMysqlSpec{Name: "mysql", Namespace: "wandb", Replicas: replicas, StorageSize: "10Gi"}, mocoWandb(), mocoScheme(), @@ -152,24 +217,52 @@ var _ = Describe("Moco MySQL specs", func() { mocoPVC := &corev1.PersistentVolumeClaim{ ObjectMeta: metav1.ObjectMeta{Name: "mysql-data-moco-mysql-0", Namespace: "wandb"}, } + credentials := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "moco-mysql", Namespace: "wandb"}, + } // A PVC matching the old (pre-Moco) "datadir-" name must NOT be matched. legacyPVC := &corev1.PersistentVolumeClaim{ ObjectMeta: metav1.ObjectMeta{Name: "datadir-mysql-0", Namespace: "wandb"}, } cl := fake.NewClientBuilder(). WithScheme(mocoScheme()). - WithObjects(mocoPVC, legacyPVC). + WithObjects(mocoPVC, legacyPVC, credentials). Build() - Expect(ensurePVCLabels(ctx, cl, "wandb", "mysql", wandbLabels)).To(Succeed()) + annotations := map[string]string{RetentionPolicyAnnotation: string(apiv2.PurgeOnDelete)} + Expect(ensurePVCMetadata(ctx, cl, "wandb", "mysql", wandbLabels, annotations)).To(Succeed()) + Expect(ensureCredentialMetadata(ctx, cl, "wandb", "mysql", wandbLabels, annotations)).To(Succeed()) got := &corev1.PersistentVolumeClaim{} Expect(cl.Get(ctx, types.NamespacedName{Name: "mysql-data-moco-mysql-0", Namespace: "wandb"}, got)).To(Succeed()) Expect(got.Labels).To(HaveKeyWithValue("app.kubernetes.io/managed-by", "wandb")) + Expect(got.Annotations).To(HaveKeyWithValue(RetentionPolicyAnnotation, string(apiv2.PurgeOnDelete))) legacy := &corev1.PersistentVolumeClaim{} Expect(cl.Get(ctx, types.NamespacedName{Name: "datadir-mysql-0", Namespace: "wandb"}, legacy)).To(Succeed()) Expect(legacy.Labels).NotTo(HaveKey("app.kubernetes.io/managed-by")) + + gotCredentials := &corev1.Secret{} + Expect(cl.Get(ctx, types.NamespacedName{Name: "moco-mysql", Namespace: "wandb"}, gotCredentials)).To(Succeed()) + Expect(gotCredentials.Labels).To(HaveKeyWithValue("app.kubernetes.io/managed-by", "wandb")) + Expect(gotCredentials.Annotations).To(HaveKeyWithValue(RetentionPolicyAnnotation, string(apiv2.PurgeOnDelete))) + }) + + It("recognizes a labeled cross-namespace cluster as attached", func() { + wandb := mocoWandb() + wandb.UID = "wandb-uid" + cluster := &mocov1beta2.MySQLCluster{ObjectMeta: metav1.ObjectMeta{ + Name: "mysql", Namespace: "database", Labels: BuildWandbMysqlLabels(wandb, "default"), + }} + cl := fake.NewClientBuilder().WithScheme(mocoScheme()).WithObjects(cluster).Build() + + Expect(CheckDetached( + context.Background(), + cl, + types.NamespacedName{Name: cluster.Name, Namespace: cluster.Namespace}, + wandb.UID, + 3, + )).To(BeNil()) }) It("backstops a scale-down the webhook can't see, leaving the cluster untouched", func() { @@ -184,6 +277,7 @@ var _ = Describe("Moco MySQL specs", func() { desired, cm, err := ToMocoMySQLClusterSpec( ctx, + apiv2.DefaultInstanceName, apiv2.ManagedMysqlSpec{Name: "mysql", Namespace: "wandb", Replicas: 1, StorageSize: "10Gi"}, mocoWandb(), mocoScheme(), diff --git a/internal/controller/infra/managed/mysql/moco/write.go b/internal/controller/infra/managed/mysql/moco/write.go index 15325fb2..ccb1f706 100644 --- a/internal/controller/infra/managed/mysql/moco/write.go +++ b/internal/controller/infra/managed/mysql/moco/write.go @@ -59,6 +59,16 @@ func WriteState( } if !found { actual = nil + } else if len(desired.OwnerReferences) == 0 { + if err := removeLegacyCrossNamespaceOwnerReferences(ctx, cl, actual); err != nil { + return []metav1.Condition{ + { + Type: common.ReconciledType, + Status: metav1.ConditionFalse, + Reason: common.ApiErrorReason, + }, + } + } } // MOCO's mutating admission webhook fills in spec.serverIDBase with a random @@ -122,6 +132,14 @@ func WriteState( } else { if !cmFound { actualConfMap = nil + } else if len(confMap.OwnerReferences) == 0 { + if cmErr := removeLegacyCrossNamespaceOwnerReferences(ctx, cl, actualConfMap); cmErr != nil { + result = append(result, metav1.Condition{ + Type: common.ReconciledType, + Status: metav1.ConditionFalse, + Reason: common.ApiErrorReason, + }) + } } if _, cmErr := common.CrudResource(ctx, cl, confMap, actualConfMap); cmErr != nil { result = append(result, metav1.Condition{ @@ -170,7 +188,17 @@ func WriteState( } if len(wandbLabels) > 0 { - if err := ensurePVCLabels(ctx, cl, specNamespacedName.Namespace, nsnBuilder.ClusterName(), wandbLabels); err != nil { + managedAnnotations := map[string]string{ + RetentionPolicyAnnotation: desired.Annotations[RetentionPolicyAnnotation], + } + if err := ensurePVCMetadata(ctx, cl, specNamespacedName.Namespace, nsnBuilder.ClusterName(), wandbLabels, managedAnnotations); err != nil { + result = append(result, metav1.Condition{ + Type: common.ReconciledType, + Status: metav1.ConditionFalse, + Reason: common.ApiErrorReason, + }) + } + if err := ensureCredentialMetadata(ctx, cl, specNamespacedName.Namespace, nsnBuilder.ClusterName(), wandbLabels, managedAnnotations); err != nil { result = append(result, metav1.Condition{ Type: common.ReconciledType, Status: metav1.ConditionFalse, @@ -182,17 +210,79 @@ func WriteState( return result } -// ensurePVCLabels stamps the wandb labels onto Moco's PVCs (missing because Moco -// doesn't propagate them through its StatefulSet volumeClaimTemplates), so -// purgeAssociatedResources can select them by label on teardown. Moco names PVCs -// "--" (see Moco pvc.go); the -// prefix is built from those same sources so it can't drift from upstream. -func ensurePVCLabels( +func removeLegacyCrossNamespaceOwnerReferences(ctx context.Context, cl client.Client, obj client.Object) error { + ownerReferences := obj.GetOwnerReferences() + filtered := make([]metav1.OwnerReference, 0, len(ownerReferences)) + for _, ref := range ownerReferences { + if ref.Kind == "WeightsAndBiases" && strings.HasPrefix(ref.APIVersion, apiv2.GroupVersion.Group+"/") { + continue + } + filtered = append(filtered, ref) + } + if len(filtered) == len(ownerReferences) { + return nil + } + patch := client.MergeFrom(obj.DeepCopyObject().(client.Object)) + obj.SetOwnerReferences(filtered) + return cl.Patch(ctx, obj, patch) +} + +func ensureCredentialMetadata( + ctx context.Context, + cl client.Client, + namespace string, + clusterName string, + desiredLabels map[string]string, + desiredAnnotations map[string]string, +) error { + secret := &corev1.Secret{} + nsName := types.NamespacedName{Namespace: namespace, Name: "moco-" + clusterName} + if err := cl.Get(ctx, nsName, secret); err != nil { + if client.IgnoreNotFound(err) == nil { + return nil + } + return err + } + if hasMetadataValues(secret, desiredLabels, desiredAnnotations) { + return nil + } + patch := client.MergeFrom(secret.DeepCopy()) + if secret.Labels == nil { + secret.Labels = make(map[string]string) + } + maps.Copy(secret.Labels, desiredLabels) + if secret.Annotations == nil { + secret.Annotations = make(map[string]string) + } + maps.Copy(secret.Annotations, desiredAnnotations) + return cl.Patch(ctx, secret, patch) +} + +func hasMetadataValues(obj client.Object, desiredLabels, desiredAnnotations map[string]string) bool { + for key, value := range desiredLabels { + if obj.GetLabels()[key] != value { + return false + } + } + for key, value := range desiredAnnotations { + if obj.GetAnnotations()[key] != value { + return false + } + } + return true +} + +// ensurePVCMetadata stamps provenance and retention metadata onto Moco's PVCs +// (Moco does not propagate it through StatefulSet volumeClaimTemplates), so +// lifecycle cleanup can select the complete instance inventory. Moco names +// PVCs "--" (see Moco pvc.go). +func ensurePVCMetadata( ctx context.Context, cl client.Client, namespace string, clusterName string, - labels map[string]string, + desiredLabels map[string]string, + desiredAnnotations map[string]string, ) error { log := logx.GetSlog(ctx) cluster := &mocov1beta2.MySQLCluster{ObjectMeta: metav1.ObjectMeta{Name: clusterName}} @@ -207,19 +297,23 @@ func ensurePVCLabels( if !strings.HasPrefix(pvc.Name, prefix) { continue } - if common.HasAllLabelKeys(pvc.Labels, labels) { + if hasMetadataValues(&pvc, desiredLabels, desiredAnnotations) { continue } patch := client.MergeFrom(pvc.DeepCopy()) if pvc.Labels == nil { pvc.Labels = make(map[string]string) } - maps.Copy(pvc.Labels, labels) + maps.Copy(pvc.Labels, desiredLabels) + if pvc.Annotations == nil { + pvc.Annotations = make(map[string]string) + } + maps.Copy(pvc.Annotations, desiredAnnotations) if err := cl.Patch(ctx, &pvc, patch); err != nil { - log.Error("failed to patch PVC labels", logx.ErrAttr(err), "pvc", pvc.Name) + log.Error("failed to patch PVC metadata", logx.ErrAttr(err), "pvc", pvc.Name) return err } - log.Debug("patched wandb labels onto PVC", "pvc", pvc.Name) + log.Debug("patched wandb metadata onto PVC", "pvc", pvc.Name) } return nil } diff --git a/internal/controller/infra/mysqlconnection/bundle.go b/internal/controller/infra/mysqlconnection/bundle.go new file mode 100644 index 00000000..81f2e84f --- /dev/null +++ b/internal/controller/infra/mysqlconnection/bundle.go @@ -0,0 +1,361 @@ +package mysqlconnection + +import ( + "context" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/hex" + "fmt" + "net" + "net/url" + "strconv" + "strings" + + apiv2 "github.com/wandb/operator/api/v2" + "github.com/wandb/operator/internal/controller/common" + 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/types" + "k8s.io/apimachinery/pkg/util/validation" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" +) + +const ( + BundleVersion = "1" + + BundleVersionAnnotation = "weightsandbiases.apps.wandb.com/mysql-bundle-version" + InstanceAnnotation = "weightsandbiases.apps.wandb.com/mysql-instance" + ProviderReadyType = "ProviderReady" + ConnectionResolvedType = "ConnectionResolved" + BundleReadyType = "BundleReady" + DatabaseInitializedType = "DatabaseInitialized" + + URLKey = "url" + HostKey = "Host" + PortKey = "Port" + DatabaseKey = "Database" + UsernameKey = "Username" + PasswordKey = "Password" + TLSKey = "Tls" + SSLCAKey = "SslCa" + SSLCertKey = "SslCert" + SSLKeyKey = "SslKey" + + CACertFile = "ca.pem" + ClientCertFile = "client-cert.pem" + ClientKeyFile = "client-key.pem" + + bundleManagedBy = "wandb-operator" + bundleComponent = "mysql-connection" +) + +const instanceIDLength = 10 + +// Material is the provider-independent MySQL connection data used to create +// the application-facing bundle. +type Material struct { + Host string + Port string + Database string + Username string + Password string + TLS string + CACert []byte + ClientCert []byte + ClientKey []byte +} + +func InstanceID(instance string) string { + if instance == "" { + instance = apiv2.DefaultInstanceName + } + digest := sha256.Sum256([]byte(instance)) + return hex.EncodeToString(digest[:])[:instanceIDLength] +} + +func SecretName(wandbName, instance string) string { + base := fmt.Sprintf("%s-%s", wandbName, InstanceID(instance)) + return common.FitDefaultInfraName(base, "-mysql-connection", validation.DNS1123LabelMaxLength) +} + +func MountPath(instance string) string { + return fmt.Sprintf("/var/run/secrets/wandb/mysql/%s", InstanceID(instance)) +} + +func VolumeName(instance string) string { + return "mysql-" + InstanceID(instance) +} + +// Normalize validates provider material and converts the v2 TLS selector value +// into the URL modes supported by the currently deployed Gorilla connector. +func Normalize(material Material) (Material, error) { + material.Host = strings.TrimSpace(material.Host) + material.Port = strings.TrimSpace(material.Port) + material.Database = strings.TrimSpace(material.Database) + material.Username = strings.TrimSpace(material.Username) + material.TLS = strings.ToLower(strings.TrimSpace(material.TLS)) + + if material.Host == "" { + return Material{}, fmt.Errorf("host is required") + } + port, err := strconv.ParseUint(material.Port, 10, 16) + if err != nil || port == 0 { + return Material{}, fmt.Errorf("port %q must be an integer between 1 and 65535", material.Port) + } + if material.Database == "" { + return Material{}, fmt.Errorf("database is required") + } + if material.Username == "" { + return Material{}, fmt.Errorf("username is required") + } + + hasCA := len(material.CACert) > 0 + hasClientCert := len(material.ClientCert) > 0 + hasClientKey := len(material.ClientKey) > 0 + if hasClientCert != hasClientKey { + return Material{}, fmt.Errorf("sslCert and sslKey must be configured together") + } + if hasClientCert && !hasCA { + return Material{}, fmt.Errorf("sslCa is required when client certificates are configured") + } + + switch material.TLS { + case "": + if hasCA { + material.TLS = "custom" + } + case "true": + if hasCA { + material.TLS = "custom" + } + case "custom": + if !hasCA { + return Material{}, fmt.Errorf("sslCa is required when tls is custom") + } + case "false": + if hasCA || hasClientCert { + return Material{}, fmt.Errorf("TLS certificates cannot be configured when tls is false") + } + case "preferred", "skip-verify": + if hasCA || hasClientCert { + return Material{}, fmt.Errorf("TLS certificates cannot be configured when tls is %s", material.TLS) + } + default: + return Material{}, fmt.Errorf("unsupported tls mode %q", material.TLS) + } + + if hasCA { + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(material.CACert) { + return Material{}, fmt.Errorf("sslCa does not contain a valid PEM certificate") + } + } + if hasClientCert { + if _, err := tls.X509KeyPair(material.ClientCert, material.ClientKey); err != nil { + return Material{}, fmt.Errorf("sslCert and sslKey are not a valid key pair: %w", err) + } + } + + return material, nil +} + +func URL(material Material, instance string) (string, error) { + normalized, err := Normalize(material) + if err != nil { + return "", err + } + + host := normalized.Host + if strings.HasPrefix(host, "[") && strings.HasSuffix(host, "]") { + host = strings.TrimSuffix(strings.TrimPrefix(host, "["), "]") + } + databaseURL := url.URL{ + Scheme: "mysql", + Host: net.JoinHostPort(host, normalized.Port), + User: url.UserPassword(normalized.Username, normalized.Password), + Path: "/" + normalized.Database, + RawPath: "/" + url.PathEscape(normalized.Database), + } + query := databaseURL.Query() + if normalized.TLS != "" { + query.Set("tls", normalized.TLS) + } + if len(normalized.CACert) > 0 { + query.Set("ssl-ca", MountPath(instance)+"/"+CACertFile) + } + if len(normalized.ClientCert) > 0 { + query.Set("ssl-cert", MountPath(instance)+"/"+ClientCertFile) + query.Set("ssl-key", MountPath(instance)+"/"+ClientKeyFile) + } + databaseURL.RawQuery = query.Encode() + return databaseURL.String(), nil +} + +func Write( + ctx context.Context, + c client.Client, + wandb *apiv2.WeightsAndBiases, + instance string, + material Material, +) (*apiv2.MysqlConnection, error) { + normalized, err := Normalize(material) + if err != nil { + return nil, err + } + connectionURL, err := URL(normalized, instance) + if err != nil { + return nil, err + } + + data := map[string][]byte{ + URLKey: []byte(connectionURL), + HostKey: []byte(normalized.Host), + PortKey: []byte(normalized.Port), + DatabaseKey: []byte(normalized.Database), + UsernameKey: []byte(normalized.Username), + PasswordKey: []byte(normalized.Password), + TLSKey: []byte(normalized.TLS), + SSLCAKey: normalized.CACert, + SSLCertKey: normalized.ClientCert, + SSLKeyKey: normalized.ClientKey, + } + + nsName := types.NamespacedName{ + Namespace: wandb.Namespace, + Name: SecretName(wandb.Name, instance), + } + actual := &corev1.Secret{} + if err := c.Get(ctx, nsName, actual); apierrors.IsNotFound(err) { + actual = nil + } else if err != nil { + return nil, err + } + + desired := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: nsName.Name, + Namespace: nsName.Namespace, + Labels: map[string]string{ + "app.kubernetes.io/managed-by": bundleManagedBy, + "app.kubernetes.io/instance": wandb.Name, + "app.kubernetes.io/component": bundleComponent, + }, + Annotations: map[string]string{ + BundleVersionAnnotation: BundleVersion, + InstanceAnnotation: instance, + }, + }, + Type: corev1.SecretTypeOpaque, + Data: data, + } + if err := controllerutil.SetOwnerReference(wandb, desired, c.Scheme()); err != nil { + return nil, err + } + if _, err := common.CrudResource(ctx, c, desired, actual); err != nil { + return nil, err + } + + return Connection(nsName.Name), nil +} + +// DeleteStale removes application bundles for instances that are no longer in +// the W&B spec. Owner UID matching prevents a newly-created W&B resource with +// the same name from deleting bundles that belong to its predecessor. +func DeleteStale( + ctx context.Context, + c client.Client, + wandb *apiv2.WeightsAndBiases, + desiredInstances map[string]struct{}, +) error { + desiredNames := make(map[string]struct{}, len(desiredInstances)) + for instance := range desiredInstances { + desiredNames[SecretName(wandb.Name, instance)] = struct{}{} + } + + secrets := &corev1.SecretList{} + if err := c.List( + ctx, + secrets, + client.InNamespace(wandb.Namespace), + client.MatchingLabels(map[string]string{ + "app.kubernetes.io/managed-by": bundleManagedBy, + "app.kubernetes.io/instance": wandb.Name, + "app.kubernetes.io/component": bundleComponent, + }), + ); err != nil { + return err + } + + for i := range secrets.Items { + secret := &secrets.Items[i] + if _, ok := desiredNames[secret.Name]; ok { + continue + } + ownedByCurrentWandb := false + for _, ref := range secret.OwnerReferences { + if ref.UID == wandb.UID { + ownedByCurrentWandb = true + break + } + } + if !ownedByCurrentWandb { + continue + } + if err := c.Delete(ctx, secret); err != nil && !apierrors.IsNotFound(err) { + return err + } + } + return nil +} + +func Read(ctx context.Context, c client.Client, wandb *apiv2.WeightsAndBiases, instance string) (*apiv2.MysqlConnection, error) { + name := SecretName(wandb.Name, instance) + secret := &corev1.Secret{} + if err := c.Get(ctx, types.NamespacedName{Namespace: wandb.Namespace, Name: name}, secret); err != nil { + if apierrors.IsNotFound(err) { + return nil, nil + } + return nil, err + } + return Connection(name), nil +} + +func Delete(ctx context.Context, c client.Client, wandb *apiv2.WeightsAndBiases, instance string) error { + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: wandb.Namespace, + Name: SecretName(wandb.Name, instance), + }, + } + if err := c.Delete(ctx, secret); err != nil && !apierrors.IsNotFound(err) { + return err + } + return nil +} + +func Connection(secretName string) *apiv2.MysqlConnection { + localRef := corev1.LocalObjectReference{Name: secretName} + selector := func(key string, optional bool) corev1.SecretKeySelector { + return corev1.SecretKeySelector{ + LocalObjectReference: localRef, + Key: key, + Optional: ptr.To(optional), + } + } + return &apiv2.MysqlConnection{ + URL: selector(URLKey, false), + Host: selector(HostKey, false), + Port: selector(PortKey, false), + Database: selector(DatabaseKey, false), + Username: selector(UsernameKey, false), + Password: selector(PasswordKey, false), + Tls: selector(TLSKey, true), + SslCa: selector(SSLCAKey, true), + SslCert: selector(SSLCertKey, true), + SslKey: selector(SSLKeyKey, true), + } +} diff --git a/internal/controller/infra/mysqlconnection/bundle_test.go b/internal/controller/infra/mysqlconnection/bundle_test.go new file mode 100644 index 00000000..28070d5a --- /dev/null +++ b/internal/controller/infra/mysqlconnection/bundle_test.go @@ -0,0 +1,332 @@ +package mysqlconnection_test + +import ( + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "net/url" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apiv2 "github.com/wandb/operator/api/v2" + "github.com/wandb/operator/internal/controller/infra/mysqlconnection" + corev1 "k8s.io/api/core/v1" + 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/fake" +) + +func TestNormalize(t *testing.T) { + caCert, clientCert, clientKey := testCertificates(t) + base := mysqlconnection.Material{ + Host: "mysql.example.com", + Port: "3306", + Database: "wandb", + Username: "wandb", + Password: "secret", + } + + tests := []struct { + name string + configure func(*mysqlconnection.Material) + wantTLS string + wantError string + }{ + {name: "plaintext", wantTLS: ""}, + { + name: "true uses driver TLS", + configure: func(material *mysqlconnection.Material) { + material.TLS = "true" + }, + wantTLS: "true", + }, + { + name: "skip verify", + configure: func(material *mysqlconnection.Material) { + material.TLS = "skip-verify" + }, + wantTLS: "skip-verify", + }, + { + name: "preferred", + configure: func(material *mysqlconnection.Material) { + material.TLS = "preferred" + }, + wantTLS: "preferred", + }, + { + name: "false disables TLS", + configure: func(material *mysqlconnection.Material) { + material.TLS = "false" + }, + wantTLS: "false", + }, + { + name: "CA enables custom TLS", + configure: func(material *mysqlconnection.Material) { + material.CACert = caCert + }, + wantTLS: "custom", + }, + { + name: "true with CA becomes custom TLS", + configure: func(material *mysqlconnection.Material) { + material.TLS = "true" + material.CACert = caCert + }, + wantTLS: "custom", + }, + { + name: "valid client certificate pair", + configure: func(material *mysqlconnection.Material) { + material.CACert = caCert + material.ClientCert = clientCert + material.ClientKey = clientKey + }, + wantTLS: "custom", + }, + { + name: "custom requires CA", + configure: func(material *mysqlconnection.Material) { + material.TLS = "custom" + }, + wantError: "sslCa is required", + }, + { + name: "client certificate requires key", + configure: func(material *mysqlconnection.Material) { + material.CACert = caCert + material.ClientCert = clientCert + }, + wantError: "configured together", + }, + { + name: "preferred rejects CA", + configure: func(material *mysqlconnection.Material) { + material.TLS = "preferred" + material.CACert = caCert + }, + wantError: "cannot be configured", + }, + { + name: "false rejects CA", + configure: func(material *mysqlconnection.Material) { + material.TLS = "false" + material.CACert = caCert + }, + wantError: "cannot be configured", + }, + { + name: "invalid CA", + configure: func(material *mysqlconnection.Material) { + material.CACert = []byte("not PEM") + }, + wantError: "valid PEM", + }, + { + name: "invalid port", + configure: func(material *mysqlconnection.Material) { + material.Port = "70000" + }, + wantError: "between 1 and 65535", + }, + { + name: "unsupported future TLS mode", + configure: func(material *mysqlconnection.Material) { + material.TLS = "verify-identity" + material.CACert = caCert + }, + wantError: "unsupported tls mode", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + material := base + if tt.configure != nil { + tt.configure(&material) + } + + got, err := mysqlconnection.Normalize(material) + if tt.wantError != "" { + require.ErrorContains(t, err, tt.wantError) + return + } + require.NoError(t, err) + assert.Equal(t, tt.wantTLS, got.TLS) + }) + } +} + +func TestURL(t *testing.T) { + caCert, _, _ := testCertificates(t) + connectionURL, err := mysqlconnection.URL(mysqlconnection.Material{ + Host: "2001:db8::1", + Port: "3306", + Database: "wandb/db", + Username: "user@example.com", + Password: "p@ss:/?#word", + CACert: caCert, + }, "analytics") + require.NoError(t, err) + + parsed, err := url.Parse(connectionURL) + require.NoError(t, err) + assert.Equal(t, "[2001:db8::1]:3306", parsed.Host) + assert.Equal(t, "user@example.com", parsed.User.Username()) + password, ok := parsed.User.Password() + require.True(t, ok) + assert.Equal(t, "p@ss:/?#word", password) + assert.Equal(t, "/wandb/db", parsed.Path) + assert.Contains(t, connectionURL, "/wandb%2Fdb") + assert.Equal(t, "custom", parsed.Query().Get("tls")) + assert.Equal( + t, + mysqlconnection.MountPath("analytics")+"/"+mysqlconnection.CACertFile, + parsed.Query().Get("ssl-ca"), + ) +} + +func TestWrite(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, apiv2.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + + wandb := &apiv2.WeightsAndBiases{ + ObjectMeta: metav1.ObjectMeta{ + Name: "wandb", + Namespace: "test", + UID: types.UID("wandb-uid"), + }, + } + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(wandb).Build() + + connection, err := mysqlconnection.Write(t.Context(), client, wandb, "analytics", mysqlconnection.Material{ + Host: "mysql.example.com", + Port: "3306", + Database: "wandb", + Username: "wandb", + Password: "secret", + }) + require.NoError(t, err) + assert.Equal(t, mysqlconnection.SecretName(wandb.Name, "analytics"), connection.URL.Name) + + secret := &corev1.Secret{} + require.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Namespace: wandb.Namespace, + Name: connection.URL.Name, + }, secret)) + assert.Equal(t, mysqlconnection.BundleVersion, secret.Annotations[mysqlconnection.BundleVersionAnnotation]) + assert.Equal(t, "secret", string(secret.Data[mysqlconnection.PasswordKey])) + for _, key := range []string{ + mysqlconnection.URLKey, + mysqlconnection.HostKey, + mysqlconnection.PortKey, + mysqlconnection.DatabaseKey, + mysqlconnection.UsernameKey, + mysqlconnection.PasswordKey, + mysqlconnection.TLSKey, + mysqlconnection.SSLCAKey, + mysqlconnection.SSLCertKey, + mysqlconnection.SSLKeyKey, + } { + assert.Contains(t, secret.Data, key) + } + assert.NotEqual( + t, + mysqlconnection.SecretName(wandb.Name, "analytics"), + mysqlconnection.SecretName(wandb.Name, "default"), + ) +} + +func TestDeleteStale(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, apiv2.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + + wandb := &apiv2.WeightsAndBiases{ + TypeMeta: metav1.TypeMeta{APIVersion: apiv2.GroupVersion.String(), Kind: "WeightsAndBiases"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "wandb", Namespace: "test", UID: types.UID("current-uid"), + }, + } + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(wandb).Build() + material := mysqlconnection.Material{ + Host: "mysql.example.com", Port: "3306", Database: "wandb", Username: "wandb", Password: "secret", + } + _, err := mysqlconnection.Write(t.Context(), client, wandb, "default", material) + require.NoError(t, err) + _, err = mysqlconnection.Write(t.Context(), client, wandb, "analytics", material) + require.NoError(t, err) + + foreign := &corev1.Secret{ObjectMeta: metav1.ObjectMeta{ + Name: "wandb-foreign-mysql-connection", Namespace: wandb.Namespace, + Labels: map[string]string{ + "app.kubernetes.io/managed-by": "wandb-operator", + "app.kubernetes.io/instance": wandb.Name, + "app.kubernetes.io/component": "mysql-connection", + }, + OwnerReferences: []metav1.OwnerReference{{UID: types.UID("old-uid")}}, + }} + require.NoError(t, client.Create(t.Context(), foreign)) + + require.NoError(t, mysqlconnection.DeleteStale( + t.Context(), client, wandb, map[string]struct{}{"default": {}}, + )) + + assert.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Namespace: wandb.Namespace, Name: mysqlconnection.SecretName(wandb.Name, "default"), + }, &corev1.Secret{})) + assert.Error(t, client.Get(t.Context(), types.NamespacedName{ + Namespace: wandb.Namespace, Name: mysqlconnection.SecretName(wandb.Name, "analytics"), + }, &corev1.Secret{})) + assert.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Namespace: wandb.Namespace, Name: foreign.Name, + }, &corev1.Secret{})) +} + +func testCertificates(t *testing.T) ([]byte, []byte, []byte) { + t.Helper() + + caKey, err := rsa.GenerateKey(rand.Reader, 2048) + require.NoError(t, err) + caTemplate := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "test CA"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + KeyUsage: x509.KeyUsageCertSign, + } + caDER, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, &caKey.PublicKey, caKey) + require.NoError(t, err) + + clientKey, err := rsa.GenerateKey(rand.Reader, 2048) + require.NoError(t, err) + clientTemplate := &x509.Certificate{ + SerialNumber: big.NewInt(2), + Subject: pkix.Name{CommonName: "client"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, + KeyUsage: x509.KeyUsageDigitalSignature, + } + clientDER, err := x509.CreateCertificate( + rand.Reader, + clientTemplate, + caTemplate, + &clientKey.PublicKey, + caKey, + ) + require.NoError(t, err) + + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER}), + pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: clientDER}), + pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(clientKey)}) +} diff --git a/internal/controller/reconciler/custom_ca.go b/internal/controller/reconciler/custom_ca.go index 8dfe6c31..9d5faed1 100644 --- a/internal/controller/reconciler/custom_ca.go +++ b/internal/controller/reconciler/custom_ca.go @@ -27,18 +27,6 @@ const ( customCACertsInlineMountPath = "/usr/local/share/ca-certificates/inline" customCACertsConfigMapMountPath = "/usr/local/share/ca-certificates/configmap" - mysqlCACertVolumeName = "mysql-ca" - mysqlCACertPath = "/etc/ssl/certs/mysql_ca.pem" - mysqlCACertFileName = "mysql_ca.pem" - - mysqlSSLCertVolumeName = "mysql-ssl-cert" - mysqlSSLCertPath = "/etc/ssl/certs/mysql_ssl_cert.pem" - mysqlSSLCertFileName = "mysql_ssl_cert.pem" - - mysqlSSLKeyVolumeName = "mysql-ssl-key" - mysqlSSLKeyPath = "/etc/ssl/certs/mysql_ssl_key.pem" - mysqlSSLKeyFileName = "mysql_ssl_key.pem" - redisCACertVolumeName = "redis-ca" redisCACertPath = "/etc/ssl/certs/redis_ca.pem" redisCACertFileName = "redis_ca.pem" @@ -58,16 +46,8 @@ func hasGlobalCustomCACertConfig(wandb *apiv2.WeightsAndBiases) bool { return len(wandb.Spec.Global.CustomCACerts) > 0 || wandb.Spec.Global.CACertsConfigMap != "" } -// defaultMySQLConnection returns the default MySQL instance's connection. The -// app's TLS env vars (MYSQL_CA_CERT_PATH etc.) are singular, so only the -// default instance's certificate material is mounted. -func defaultMySQLConnection(wandb *apiv2.WeightsAndBiases) apiv2.MysqlConnection { - status, _ := apiv2.ResolveInstance(wandb.Status.MySQLStatus, "") - return status.Connection -} - // defaultRedisConnection returns the default Redis instance's connection; see -// defaultMySQLConnection. +// the legacy singular Redis CA wiring. func defaultRedisConnection(wandb *apiv2.WeightsAndBiases) apiv2.RedisConnection { status, _ := apiv2.ResolveInstance(wandb.Status.RedisStatus, "") return status.Connection @@ -141,7 +121,6 @@ func applyCustomCACertsToWorkload( volumes []corev1.Volume, volumeMounts []corev1.VolumeMount, ) ([]corev1.EnvVar, []corev1.Volume, []corev1.VolumeMount, string, error) { - mysqlConn := defaultMySQLConnection(wandb) redisConn := defaultRedisConnection(wandb) if hasGlobalCustomCACertConfig(wandb) { @@ -192,56 +171,6 @@ func applyCustomCACertsToWorkload( } } - if hasValue, err := secretSelectorHasValue(ctx, c, wandb.Namespace, mysqlConn.SslCa); err != nil { - return nil, nil, nil, "", err - } else if hasValue { - envVars = appendMissingEnvVars(envVars, []corev1.EnvVar{{Name: "MYSQL_CA_CERT_PATH", Value: mysqlCACertPath}}) - volumes = upsertVolume(volumes, corev1.Volume{ - Name: mysqlCACertVolumeName, - VolumeSource: corev1.VolumeSource{ - Secret: secretCACertVolumeSource(mysqlConn.SslCa, mysqlCACertFileName), - }, - }) - volumeMounts = upsertVolumeMount(volumeMounts, corev1.VolumeMount{ - Name: mysqlCACertVolumeName, - MountPath: mysqlCACertPath, - SubPath: mysqlCACertFileName, - ReadOnly: true, - }) - } - if hasValue, err := secretSelectorHasValue(ctx, c, wandb.Namespace, mysqlConn.SslCert); err != nil { - return nil, nil, nil, "", err - } else if hasValue { - volumes = upsertVolume(volumes, corev1.Volume{ - Name: mysqlSSLCertVolumeName, - VolumeSource: corev1.VolumeSource{ - Secret: secretCACertVolumeSource(mysqlConn.SslCert, mysqlSSLCertFileName), - }, - }) - volumeMounts = upsertVolumeMount(volumeMounts, corev1.VolumeMount{ - Name: mysqlSSLCertVolumeName, - MountPath: mysqlSSLCertPath, - SubPath: mysqlSSLCertFileName, - ReadOnly: true, - }) - } - if hasValue, err := secretSelectorHasValue(ctx, c, wandb.Namespace, mysqlConn.SslKey); err != nil { - return nil, nil, nil, "", err - } else if hasValue { - volumes = upsertVolume(volumes, corev1.Volume{ - Name: mysqlSSLKeyVolumeName, - VolumeSource: corev1.VolumeSource{ - Secret: secretCACertVolumeSource(mysqlConn.SslKey, mysqlSSLKeyFileName), - }, - }) - volumeMounts = upsertVolumeMount(volumeMounts, corev1.VolumeMount{ - Name: mysqlSSLKeyVolumeName, - MountPath: mysqlSSLKeyPath, - SubPath: mysqlSSLKeyFileName, - ReadOnly: true, - }) - } - if hasValue, err := secretSelectorHasValue(ctx, c, wandb.Namespace, redisConn.SslCa); err != nil { return nil, nil, nil, "", err } else if hasValue { @@ -341,27 +270,14 @@ func secretSelectorHasValue(ctx context.Context, c ctrlClient.Client, namespace } func customCACertsChecksum(ctx context.Context, c ctrlClient.Client, wandb *apiv2.WeightsAndBiases) (string, error) { - mysqlConn := defaultMySQLConnection(wandb) redisConn := defaultRedisConnection(wandb) - hasMySQLCA, err := secretSelectorHasValue(ctx, c, wandb.Namespace, mysqlConn.SslCa) - if err != nil { - return "", err - } - hasMySQLCert, err := secretSelectorHasValue(ctx, c, wandb.Namespace, mysqlConn.SslCert) - if err != nil { - return "", err - } - hasMySQLKey, err := secretSelectorHasValue(ctx, c, wandb.Namespace, mysqlConn.SslKey) - if err != nil { - return "", err - } hasRedisCA, err := secretSelectorHasValue(ctx, c, wandb.Namespace, redisConn.SslCa) if err != nil { return "", err } - if !hasGlobalCustomCACertConfig(wandb) && !hasMySQLCA && !hasMySQLCert && !hasMySQLKey && !hasRedisCA { + if !hasGlobalCustomCACertConfig(wandb) && !hasRedisCA { return "", nil } @@ -377,25 +293,6 @@ func customCACertsChecksum(ctx context.Context, c ctrlClient.Client, wandb *apiv } } - if sel := mysqlConn.SslCa; hasMySQLCA { - _, _ = fmt.Fprintf(hash, "mysql:%s/%s\n", sel.Name, sel.Key) - if err := hashSecretKeyData(ctx, c, wandb.Namespace, sel, hashWriteString(hash)); err != nil { - return "", err - } - } - if sel := mysqlConn.SslCert; hasMySQLCert { - _, _ = fmt.Fprintf(hash, "mysql-cert:%s/%s\n", sel.Name, sel.Key) - if err := hashSecretKeyData(ctx, c, wandb.Namespace, sel, hashWriteString(hash)); err != nil { - return "", err - } - } - if sel := mysqlConn.SslKey; hasMySQLKey { - _, _ = fmt.Fprintf(hash, "mysql-key:%s/%s\n", sel.Name, sel.Key) - if err := hashSecretKeyData(ctx, c, wandb.Namespace, sel, hashWriteString(hash)); err != nil { - return "", err - } - } - if sel := redisConn.SslCa; hasRedisCA { _, _ = fmt.Fprintf(hash, "redis:%s/%s\n", sel.Name, sel.Key) if err := hashSecretKeyData(ctx, c, wandb.Namespace, sel, hashWriteString(hash)); err != nil { diff --git a/internal/controller/reconciler/custom_ca_test.go b/internal/controller/reconciler/custom_ca_test.go index ea43e3df..71cb668a 100644 --- a/internal/controller/reconciler/custom_ca_test.go +++ b/internal/controller/reconciler/custom_ca_test.go @@ -92,24 +92,6 @@ func TestApplyCustomCACertsToWorkloadAddsGlobalAndInfraMounts(t *testing.T) { }, }, Status: apiv2.WeightsAndBiasesStatus{ - MySQLStatus: map[string]apiv2.MysqlInfraStatus{ - apiv2.DefaultInstanceName: apiv2.MysqlInfraStatus{ - Connection: apiv2.MysqlConnection{ - SslCa: corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{Name: "wandb-mysql-connection"}, - Key: "SslCa", - }, - SslCert: corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{Name: "wandb-mysql-connection"}, - Key: "SslCert", - }, - SslKey: corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{Name: "wandb-mysql-connection"}, - Key: "SslKey", - }, - }, - }, - }, RedisStatus: map[string]apiv2.RedisInfraStatus{ apiv2.DefaultInstanceName: apiv2.RedisInfraStatus{ Connection: apiv2.RedisConnection{ @@ -123,14 +105,6 @@ func TestApplyCustomCACertsToWorkloadAddsGlobalAndInfraMounts(t *testing.T) { }, }, } - mysqlSecret := &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{Name: "wandb-mysql-connection", Namespace: "default"}, - Data: map[string][]byte{ - "SslCa": []byte("---mysql-ca---"), - "SslCert": []byte("---mysql-cert---"), - "SslKey": []byte("---mysql-key---"), - }, - } redisSecret := &corev1.Secret{ ObjectMeta: metav1.ObjectMeta{Name: "wandb-redis-connection", Namespace: "default"}, Data: map[string][]byte{"SslCa": []byte("---redis-ca---")}, @@ -139,7 +113,7 @@ func TestApplyCustomCACertsToWorkloadAddsGlobalAndInfraMounts(t *testing.T) { ObjectMeta: metav1.ObjectMeta{Name: "user-ca-certs", Namespace: "default"}, Data: map[string]string{"corp.crt": "---corp---"}, } - builder := customCATestClient(t, wandb, mysqlSecret, redisSecret, userCM) + builder := customCATestClient(t, wandb, redisSecret, userCM) client := builder.Build() envs, volumes, mounts, checksum, err := applyCustomCACertsToWorkload(context.Background(), client, wandb, nil, nil, nil) @@ -149,22 +123,15 @@ func TestApplyCustomCACertsToWorkloadAddsGlobalAndInfraMounts(t *testing.T) { requireContainsEnv(t, envs, "SSL_CERT_FILE", "/etc/ssl/certs/ca-certificates.crt") requireContainsEnv(t, envs, "SSL_CERT_DIR", "/etc/ssl/certs") requireContainsEnv(t, envs, "REQUESTS_CA_BUNDLE", "/etc/ssl/certs/ca-certificates.crt") - requireContainsEnv(t, envs, "MYSQL_CA_CERT_PATH", mysqlCACertPath) requireVolume(t, volumes, customCACertsRootVolumeName) requireVolume(t, volumes, customCACertsInlineVolumeName) requireVolume(t, volumes, customCACertsConfigMapVolumeName) - requireVolume(t, volumes, mysqlCACertVolumeName) - requireVolume(t, volumes, mysqlSSLCertVolumeName) - requireVolume(t, volumes, mysqlSSLKeyVolumeName) requireVolume(t, volumes, redisCACertVolumeName) requireMount(t, mounts, customCACertsRootVolumeName, customCACertsRootMountPath) requireMount(t, mounts, customCACertsInlineVolumeName, customCACertsInlineMountPath) requireMount(t, mounts, customCACertsConfigMapVolumeName, customCACertsConfigMapMountPath) - requireMount(t, mounts, mysqlCACertVolumeName, mysqlCACertPath) - requireMount(t, mounts, mysqlSSLCertVolumeName, mysqlSSLCertPath) - requireMount(t, mounts, mysqlSSLKeyVolumeName, mysqlSSLKeyPath) requireMount(t, mounts, redisCACertVolumeName, redisCACertPath) podTemplate := &corev1.PodTemplateSpec{} @@ -179,17 +146,6 @@ func TestApplyCustomCACertsToWorkloadSkipsMissingOptionalInfraKeys(t *testing.T) Namespace: "default", }, Status: apiv2.WeightsAndBiasesStatus{ - MySQLStatus: map[string]apiv2.MysqlInfraStatus{ - apiv2.DefaultInstanceName: apiv2.MysqlInfraStatus{ - Connection: apiv2.MysqlConnection{ - SslCa: corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{Name: "wandb-mysql-connection"}, - Key: "SslCa", - Optional: ptr.To(true), - }, - }, - }, - }, RedisStatus: map[string]apiv2.RedisInfraStatus{ apiv2.DefaultInstanceName: apiv2.RedisInfraStatus{ Connection: apiv2.RedisConnection{ @@ -203,22 +159,16 @@ func TestApplyCustomCACertsToWorkloadSkipsMissingOptionalInfraKeys(t *testing.T) }, }, } - mysqlSecret := &corev1.Secret{ - ObjectMeta: metav1.ObjectMeta{Name: "wandb-mysql-connection", Namespace: "default"}, - Data: map[string][]byte{"url": []byte("mysql://user:pass@db:3306/wandb")}, - } redisSecret := &corev1.Secret{ ObjectMeta: metav1.ObjectMeta{Name: "wandb-redis-connection", Namespace: "default"}, Data: map[string][]byte{"url": []byte("redis://redis:6379")}, } - builder := customCATestClient(t, wandb, mysqlSecret, redisSecret) + builder := customCATestClient(t, wandb, redisSecret) client := builder.Build() - envs, volumes, mounts, checksum, err := applyCustomCACertsToWorkload(context.Background(), client, wandb, nil, nil, nil) + _, volumes, mounts, checksum, err := applyCustomCACertsToWorkload(context.Background(), client, wandb, nil, nil, nil) require.NoError(t, err) require.Empty(t, checksum) - requireNoEnv(t, envs, "MYSQL_CA_CERT_PATH") - requireNoVolume(t, volumes, mysqlCACertVolumeName) requireNoVolume(t, volumes, redisCACertVolumeName) require.Empty(t, mounts) } @@ -234,13 +184,6 @@ func requireContainsEnv(t *testing.T, envs []corev1.EnvVar, name, value string) t.Fatalf("env var %q not found in %+v", name, envs) } -func requireNoEnv(t *testing.T, envs []corev1.EnvVar, name string) { - t.Helper() - for _, env := range envs { - require.NotEqual(t, name, env.Name) - } -} - func requireVolume(t *testing.T, volumes []corev1.Volume, name string) { t.Helper() for _, volume := range volumes { diff --git a/internal/controller/reconciler/mysql.go b/internal/controller/reconciler/mysql.go index a078e521..541150df 100644 --- a/internal/controller/reconciler/mysql.go +++ b/internal/controller/reconciler/mysql.go @@ -10,12 +10,14 @@ import ( "github.com/wandb/operator/internal/controller/infra/external" externalmysql "github.com/wandb/operator/internal/controller/infra/external/mysql" "github.com/wandb/operator/internal/controller/infra/managed/mysql/moco" + "github.com/wandb/operator/internal/controller/infra/mysqlconnection" "github.com/wandb/operator/pkg/utils" "github.com/wandb/operator/pkg/wandb/manifest" "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/tools/record" ctrl "sigs.k8s.io/controller-runtime" @@ -28,17 +30,101 @@ func mysqlWriteState( client client.Client, wandb *apiv2.WeightsAndBiases, mfst manifest.Manifest, -) map[string][]metav1.Condition { +) (map[string][]metav1.Condition, error) { + if err := reconcileStaleManagedMySQL(ctx, client, wandb); err != nil { + return nil, err + } + desiredInstances := make(map[string]struct{}, len(wandb.Spec.MySQL)) + for key := range wandb.Spec.MySQL { + desiredInstances[key] = struct{}{} + } + if err := mysqlconnection.DeleteStale(ctx, client, wandb, desiredInstances); err != nil { + return nil, err + } out := map[string][]metav1.Condition{} for key, spec := range wandb.Spec.MySQL { switch { case spec.ManagedMysql != nil: - out[key] = managedMysqlWriteState(ctx, client, wandb, spec.ManagedMysql, mfst) + out[key] = managedMysqlWriteState(ctx, client, wandb, key, spec.ManagedMysql, mfst) case spec.ExternalMysql != nil: out[key] = externalmysql.WriteState(ctx, client, wandb, key, spec.ExternalMysql) } } - return out + return out, nil +} + +func reconcileStaleManagedMySQL(ctx context.Context, c client.Client, wandb *apiv2.WeightsAndBiases) error { + // Keyed by the Moco resource each desired instance points at: an instance key + // can be renamed while still referring to the same live database. + desired := map[types.NamespacedName]struct{}{} + for _, spec := range wandb.Spec.MySQL { + if spec.ManagedMysql == nil { + continue + } + desired[managedMysqlSpecNamespacedName(spec.ManagedMysql)] = struct{}{} + } + + clusters := &mocov1beta2.MySQLClusterList{} + if err := c.List(ctx, clusters, client.MatchingLabels(common.BuildWandbLabels(wandb, moco.MysqlModuleName))); err != nil { + return err + } + for i := range clusters.Items { + cluster := &clusters.Items[i] + ownerMatches := !common.IsDetached(cluster, wandb.UID) + labelUID := cluster.Labels[moco.WandbUIDLabel] + if labelUID != "" && labelUID != string(wandb.UID) { + continue + } + if labelUID == "" && !ownerMatches { + continue + } + + clusterName := client.ObjectKeyFromObject(cluster) + if _, ok := desired[clusterName]; ok { + // Still wanted: undo a detach this deployment applied earlier so a + // removed-then-restored instance does not stay frozen forever. + if labelUID == string(wandb.UID) { + if err := moco.ClearDetached(ctx, c, clusterName); err != nil { + return err + } + } + continue + } + if cluster.Annotations[moco.DetachedAnnotation] == "true" { + continue + } + + policy := apiv2.OnDeletePolicy(cluster.Annotations[moco.RetentionPolicyAnnotation]) + if policy == "" { + policy = wandb.Spec.RetentionPolicy.OnDelete + } + if policy == apiv2.PurgeOnDelete { + // Only select resources provably belonging to this instance: the + // deployment-wide labels alone also match sibling instances, and a + // cluster predating the instance labels has nothing narrower. The + // cluster's own storage is purged by name instead. + selector := labels.Nothing() + if instanceID := cluster.Labels[moco.InstanceLabel]; instanceID != "" { + selectorLabels := common.BuildWandbLabels(wandb, moco.MysqlModuleName) + selectorLabels[moco.InstanceLabel] = instanceID + if labelUID != "" { + selectorLabels[moco.WandbUIDLabel] = labelUID + } + selector = labels.SelectorFromSet(selectorLabels) + } + if err := moco.PurgeFinalizer(ctx, c, clusterName, common.OnDeleteRule{ + Policy: common.Purge, + Selector: selector, + }); err != nil { + return err + } + continue + } + if err := moco.DetachFinalizer(ctx, c, clusterName, wandb); err != nil { + return err + } + } + return nil } func mysqlReadState( @@ -52,7 +138,7 @@ func mysqlReadState( for key, spec := range wandb.Spec.MySQL { switch { case spec.ManagedMysql != nil: - outConds[key], outConns[key] = managedMysqlReadState(ctx, client, wandb, spec.ManagedMysql, conditions[key]) + outConds[key], outConns[key] = managedMysqlReadState(ctx, client, wandb, key, spec.ManagedMysql, conditions[key]) case spec.ExternalMysql != nil: outConds[key], outConns[key] = externalmysql.ReadState(ctx, client, wandb, key, conditions[key]) default: @@ -120,7 +206,7 @@ func mysqlPurgeFinalizer( ) error { if managed := spec.ManagedMysql; managed != nil { specNamespacedName := managedMysqlSpecNamespacedName(managed) - onDeleteRule := moco.ToMysqlOnDeleteRule(wandb, wandb.GetRetentionPolicy(managed.ManagedInfraSpec)) + onDeleteRule := moco.ToMysqlOnDeleteRule(wandb, key, wandb.GetRetentionPolicy(managed.ManagedInfraSpec)) return moco.PurgeFinalizer(ctx, client, specNamespacedName, onDeleteRule) } if spec.ExternalMysql != nil { @@ -150,77 +236,24 @@ func managedMysqlWriteState( ctx context.Context, client client.Client, wandb *apiv2.WeightsAndBiases, + key string, spec *apiv2.ManagedMysqlSpec, mfst manifest.Manifest, ) []metav1.Condition { var specNamespacedName = managedMysqlSpecNamespacedName(spec) logger := ctrl.LoggerFrom(ctx) - dbPasswordSecret := &corev1.Secret{} - err := client.Get(ctx, types.NamespacedName{Name: fmt.Sprintf("%s-%s", specNamespacedName.Name, "db-password"), Namespace: specNamespacedName.Namespace}, dbPasswordSecret) - if err != nil { - if errors.IsNotFound(err) { - dbPasswordSecret.Name = fmt.Sprintf("%s-%s", specNamespacedName.Name, "db-password") - dbPasswordSecret.Namespace = specNamespacedName.Namespace - userPassword, err := utils.GenerateRandomPassword(32) - if err != nil { - logger.Error(err, "failed to generate random password") - return []metav1.Condition{ - { - Type: common.ReconciledType, - Status: metav1.ConditionFalse, - Reason: common.ControllerErrorReason, - }, - } - } - rootPassword, err := utils.GenerateRandomPassword(32) - if err != nil { - logger.Error(err, "failed to generate random password") - return []metav1.Condition{ - { - Type: common.ReconciledType, - Status: metav1.ConditionFalse, - Reason: common.ControllerErrorReason, - }, - } - } - - dbPasswordSecret.Labels = moco.BuildWandbMysqlLabels(wandb) - dbPasswordSecret.Data = map[string][]byte{ - "rootUser": []byte("root"), - "rootPassword": []byte(rootPassword), - "rootHost": []byte("%"), - "password": []byte(userPassword), - } - if err = client.Create(ctx, dbPasswordSecret); err != nil { - logger.Error(err, "failed to create db password secret") - return []metav1.Condition{ - { - Type: common.ReconciledType, - Status: metav1.ConditionFalse, - Reason: common.ApiErrorReason, - }, - } - } - } else { - logger.Error(err, "failed to retrieve db password secret") - return []metav1.Condition{ - { - Type: common.ReconciledType, - Status: metav1.ConditionFalse, - Reason: common.ApiErrorReason, - }, - } - } - } - if conditions := moco.CheckDetached(ctx, client, specNamespacedName, wandb.GetUID(), spec.Replicas); conditions != nil { - return conditions + return append(conditions, metav1.Condition{ + Type: mysqlconnection.ProviderReadyType, + Status: metav1.ConditionFalse, + Reason: "Detached", + }) } var desired *mocov1beta2.MySQLCluster var confMap *corev1.ConfigMap - desired, confMap, err = moco.ToMocoMySQLClusterSpec(ctx, *spec, wandb, client.Scheme(), mfst) + desired, confMap, err := moco.ToMocoMySQLClusterSpec(ctx, key, *spec, wandb, client.Scheme(), mfst) if err != nil { logger.Error(err, "failed to translate moco spec") return []metav1.Condition{ @@ -229,23 +262,77 @@ func managedMysqlWriteState( Status: metav1.ConditionFalse, Reason: common.ControllerErrorReason, }, + { + Type: mysqlconnection.ProviderReadyType, + Status: metav1.ConditionFalse, + Reason: common.ControllerErrorReason, + }, } } - return moco.WriteState(ctx, client, specNamespacedName, desired, confMap, moco.BuildWandbMysqlLabels(wandb)) + conditions := moco.WriteState(ctx, client, specNamespacedName, desired, confMap, moco.BuildWandbMysqlLabels(wandb, key)) + providerCondition := metav1.Condition{ + Type: mysqlconnection.ProviderReadyType, + Status: metav1.ConditionFalse, + Reason: "Provisioning", + } + for _, condition := range conditions { + if condition.Type == moco.MySQLCustomResourceType { + providerCondition.Status = condition.Status + providerCondition.Reason = condition.Reason + } + if condition.Type == common.ReconciledType && condition.Status == metav1.ConditionFalse { + providerCondition.Status = metav1.ConditionFalse + providerCondition.Reason = condition.Reason + } + } + return append(conditions, providerCondition) } func managedMysqlReadState( ctx context.Context, client client.Client, wandb *apiv2.WeightsAndBiases, + key string, spec *apiv2.ManagedMysqlSpec, newConditions []metav1.Condition, ) ([]metav1.Condition, *apiv2.MysqlConnection) { specNamespacedName := managedMysqlSpecNamespacedName(spec) - readConditions, newInfraConn := moco.ReadState(ctx, client, specNamespacedName, wandb, moco.ToMysqlOnDeleteRule(wandb, wandb.GetRetentionPolicy(spec.ManagedInfraSpec))) + readConditions, material := moco.ReadState(ctx, client, specNamespacedName, moco.ToMysqlOnDeleteRule(wandb, key, wandb.GetRetentionPolicy(spec.ManagedInfraSpec))) newConditions = append(newConditions, readConditions...) - return newConditions, newInfraConn + if material == nil { + return newConditions, nil + } + + connection, err := mysqlconnection.Write(ctx, client, wandb, key, *material) + if err != nil { + return append(newConditions, + metav1.Condition{ + Type: mysqlconnection.BundleReadyType, + Status: metav1.ConditionFalse, + Reason: common.ApiErrorReason, + Message: err.Error(), + }, + metav1.Condition{ + Type: moco.MySQLConnectionInfoType, + Status: metav1.ConditionFalse, + Reason: common.ApiErrorReason, + }, + ), nil + } + newConditions = append(newConditions, + metav1.Condition{ + Type: mysqlconnection.ConnectionResolvedType, + Status: metav1.ConditionTrue, + Reason: "MocoCredentialsResolved", + }, + metav1.Condition{ + Type: mysqlconnection.BundleReadyType, + Status: metav1.ConditionTrue, + Reason: "BundleWritten", + }, + ) + return newConditions, connection } func managedMysqlInferStatus( @@ -262,6 +349,19 @@ func managedMysqlInferStatus( oldStatus := wandb.Status.MySQLStatus[key] oldConditions := oldStatus.Conditions oldInfraConn := oldStatus.Connection + initStatus := wandb.Status.Wandb.MySQLInit[key] + initialized := metav1.Condition{ + Type: mysqlconnection.DatabaseInitializedType, + Status: metav1.ConditionFalse, + Reason: "InitializationPending", + } + if initStatus.Succeeded { + initialized.Status = metav1.ConditionTrue + initialized.Reason = "JobSucceeded" + } else if initStatus.Failed { + initialized.Reason = "JobFailed" + } + newConditions = append(newConditions, initialized) updatedStatus, events, ctrlResult := moco.ComputeStatus( ctx, @@ -363,26 +463,52 @@ func runMysqlInitJobInstance(ctx context.Context, client client.Client, wandb *a if err != nil && !errors.IsNotFound(err) { return ctrl.Result{}, err } + jobNotFound := errors.IsNotFound(err) + _, _, bundleChecksum, err := applyMySQLBundlesToWorkload( + ctx, + client, + wandb, + map[string]struct{}{key: {}}, + nil, + nil, + ) + if err != nil { + return ctrl.Result{}, err + } + if !jobNotFound && job.Status.Succeeded == 0 && bundleChecksum != "" && + job.Spec.Template.Annotations[mysqlBundlesChecksumAnnotation] != bundleChecksum { + if err := deleteJobCascading(ctx, client, job); err != nil && !errors.IsNotFound(err) { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: defaultRequeueDuration}, nil + } - if errors.IsNotFound(err) { + if jobNotFound { logger.Info("Creating MySQL init job") - connSecretName := fmt.Sprintf("%s-connection", specNamespacedName.Name) - // moco-writable has DDL/DML privileges on all non-system databases, // so CREATE DATABASE works. The Oracle-era CREATE USER + GRANT steps // are unnecessary — wandb connects directly as the secret's Username. - mysqlCmd := `mysql -h "$MYSQL_HOST" -P "$MYSQL_PORT" -u "$MYSQL_USER" -p"$MYSQL_PWD" ` + - `-e "CREATE DATABASE IF NOT EXISTS $MYSQL_DB;"` + connectionStatus, ok := wandb.Status.MySQLStatus[key] + if !ok || connectionStatus.Connection.URL.Name == "" { + return ctrl.Result{RequeueAfter: defaultRequeueDuration}, nil + } + connection := connectionStatus.Connection + selectorOrDefault := func(selector corev1.SecretKeySelector, bundleKey string) corev1.SecretKeySelector { + if selector.Name != "" && selector.Key != "" { + return selector + } + return corev1.SecretKeySelector{ + LocalObjectReference: connection.URL.LocalObjectReference, + Key: bundleKey, + } + } - envFromConn := func(name, key string) corev1.EnvVar { + envFromConn := func(name string, selector corev1.SecretKeySelector) corev1.EnvVar { return corev1.EnvVar{ Name: name, ValueFrom: &corev1.EnvVarSource{ - SecretKeyRef: &corev1.SecretKeySelector{ - LocalObjectReference: corev1.LocalObjectReference{Name: connSecretName}, - Key: key, - }, + SecretKeyRef: &selector, }, } } @@ -405,13 +531,18 @@ func runMysqlInitJobInstance(ctx context.Context, client client.Client, wandb *a { Name: "moco-init", Image: moco.MocoMySQLImage(mysqlManifestConfig(mfst, key).Images["mysql"], wandb.Spec.Global.ImageRegistry), - Command: []string{"/bin/sh", "-c", mysqlCmd}, + Command: []string{"mysql"}, + Args: []string{ + "--host=$(MYSQL_HOST)", + "--port=$(MYSQL_PORT)", + "--user=$(MYSQL_USER)", + "--execute=CREATE DATABASE IF NOT EXISTS `wandb_local`;", + }, Env: []corev1.EnvVar{ - envFromConn("MYSQL_HOST", "Host"), - envFromConn("MYSQL_PORT", "Port"), - envFromConn("MYSQL_USER", "Username"), - envFromConn("MYSQL_PWD", "Password"), - envFromConn("MYSQL_DB", "Database"), + envFromConn("MYSQL_HOST", selectorOrDefault(connection.Host, mysqlconnection.HostKey)), + envFromConn("MYSQL_PORT", selectorOrDefault(connection.Port, mysqlconnection.PortKey)), + envFromConn("MYSQL_USER", selectorOrDefault(connection.Username, mysqlconnection.UsernameKey)), + envFromConn("MYSQL_PWD", selectorOrDefault(connection.Password, mysqlconnection.PasswordKey)), }, }, }, @@ -423,6 +554,7 @@ func runMysqlInitJobInstance(ctx context.Context, client client.Client, wandb *a if err := controllerutil.SetOwnerReference(wandb, job, client.Scheme()); err != nil { return ctrl.Result{}, err } + setMySQLBundlesChecksumAnnotation(&job.Spec.Template, bundleChecksum) if err := client.Create(ctx, job); err != nil { return ctrl.Result{}, err diff --git a/internal/controller/reconciler/mysql_bundles.go b/internal/controller/reconciler/mysql_bundles.go new file mode 100644 index 00000000..ce4ac001 --- /dev/null +++ b/internal/controller/reconciler/mysql_bundles.go @@ -0,0 +1,142 @@ +package reconciler + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "sort" + + apiv2 "github.com/wandb/operator/api/v2" + "github.com/wandb/operator/internal/controller/infra/mysqlconnection" + batchv1 "k8s.io/api/batch/v1" + 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/types" + ctrlclient "sigs.k8s.io/controller-runtime/pkg/client" +) + +const mysqlBundlesChecksumAnnotation = "weightsandbiases.apps.wandb.com/mysql-bundles-checksum" + +func applyMySQLBundlesToWorkload( + ctx context.Context, + c ctrlclient.Client, + wandb *apiv2.WeightsAndBiases, + instances map[string]struct{}, + volumes []corev1.Volume, + volumeMounts []corev1.VolumeMount, +) ([]corev1.Volume, []corev1.VolumeMount, string, error) { + if len(instances) == 0 { + return volumes, volumeMounts, "", nil + } + + keys := make([]string, 0, len(instances)) + for key := range instances { + keys = append(keys, key) + } + sort.Strings(keys) + + hash := sha256.New() + hasBundles := false + for _, key := range keys { + status, ok := wandb.Status.MySQLStatus[key] + if !ok || status.Connection.URL.Name == "" { + // Status has not caught up with the spec yet; mount what exists and + // let the checksum change once the bundle is published. + continue + } + + secret := &corev1.Secret{} + nsName := types.NamespacedName{ + Namespace: wandb.Namespace, + Name: status.Connection.URL.Name, + } + if err := c.Get(ctx, nsName, secret); apierrors.IsNotFound(err) { + // Status can briefly refer to a legacy connection Secret during an + // operator upgrade. The provider reconcile replaces it before new + // bundle mounts are required. + continue + } else if err != nil { + return nil, nil, "", fmt.Errorf("read mysql bundle %s: %w", nsName, err) + } + if secret.Annotations[mysqlconnection.BundleVersionAnnotation] == "" { + continue + } + hasBundles = true + + _, _ = fmt.Fprintf(hash, "instance:%s\nsecret:%s\n", key, secret.Name) + dataKeys := make([]string, 0, len(secret.Data)) + for dataKey := range secret.Data { + dataKeys = append(dataKeys, dataKey) + } + sort.Strings(dataKeys) + for _, dataKey := range dataKeys { + _, _ = fmt.Fprintf(hash, "%s=", dataKey) + _, _ = hash.Write(secret.Data[dataKey]) + _, _ = hash.Write([]byte("\n")) + } + + items := make([]corev1.KeyToPath, 0, 3) + if len(secret.Data[mysqlconnection.SSLCAKey]) > 0 { + items = append(items, corev1.KeyToPath{Key: mysqlconnection.SSLCAKey, Path: mysqlconnection.CACertFile}) + } + if len(secret.Data[mysqlconnection.SSLCertKey]) > 0 { + items = append(items, + corev1.KeyToPath{Key: mysqlconnection.SSLCertKey, Path: mysqlconnection.ClientCertFile}, + corev1.KeyToPath{Key: mysqlconnection.SSLKeyKey, Path: mysqlconnection.ClientKeyFile}, + ) + } + if len(items) == 0 { + continue + } + + volumeName := mysqlconnection.VolumeName(key) + volumes = upsertVolume(volumes, corev1.Volume{ + Name: volumeName, + VolumeSource: corev1.VolumeSource{ + Secret: &corev1.SecretVolumeSource{ + SecretName: secret.Name, + Items: items, + }, + }, + }) + volumeMounts = upsertVolumeMount(volumeMounts, corev1.VolumeMount{ + Name: volumeName, + MountPath: mysqlconnection.MountPath(key), + ReadOnly: true, + }) + } + if !hasBundles { + return volumes, volumeMounts, "", nil + } + + return volumes, volumeMounts, hex.EncodeToString(hash.Sum(nil)), nil +} + +// deleteJobCascading deletes a Job together with its pods: a superseded pod left +// running would race the pod of the recreated Job against the same database. +func deleteJobCascading(ctx context.Context, c ctrlclient.Client, job *batchv1.Job) error { + propagation := metav1.DeletePropagationBackground + return c.Delete(ctx, job, &ctrlclient.DeleteOptions{PropagationPolicy: &propagation}) +} + +func setMySQLBundlesChecksumAnnotation(podTemplate *corev1.PodTemplateSpec, checksum string) { + annotations := podTemplate.GetAnnotations() + if checksum == "" { + if annotations == nil { + return + } + delete(annotations, mysqlBundlesChecksumAnnotation) + if len(annotations) == 0 { + annotations = nil + } + podTemplate.SetAnnotations(annotations) + return + } + if annotations == nil { + annotations = map[string]string{} + } + annotations[mysqlBundlesChecksumAnnotation] = checksum + podTemplate.SetAnnotations(annotations) +} diff --git a/internal/controller/reconciler/mysql_bundles_test.go b/internal/controller/reconciler/mysql_bundles_test.go new file mode 100644 index 00000000..d633beb6 --- /dev/null +++ b/internal/controller/reconciler/mysql_bundles_test.go @@ -0,0 +1,118 @@ +package reconciler + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apiv2 "github.com/wandb/operator/api/v2" + "github.com/wandb/operator/internal/controller/infra/mysqlconnection" + corev1 "k8s.io/api/core/v1" + 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/fake" +) + +func TestApplyMySQLBundlesToWorkloadScopesMountsAndDigest(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + + defaultSecret := mysqlBundleTestSecret("default-conn", "default-password") + analyticsSecret := mysqlBundleTestSecret("analytics-conn", "analytics-password") + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(defaultSecret, analyticsSecret).Build() + wandb := &apiv2.WeightsAndBiases{ + ObjectMeta: metav1.ObjectMeta{Name: "wandb", Namespace: "default"}, + Status: apiv2.WeightsAndBiasesStatus{MySQLStatus: map[string]apiv2.MysqlInfraStatus{ + apiv2.DefaultInstanceName: {Connection: *mysqlconnection.Connection(defaultSecret.Name)}, + "analytics": {Connection: *mysqlconnection.Connection(analyticsSecret.Name)}, + }}, + } + + volumes, mounts, firstDigest, err := applyMySQLBundlesToWorkload( + t.Context(), + client, + wandb, + map[string]struct{}{"analytics": {}}, + nil, + nil, + ) + require.NoError(t, err) + require.Len(t, volumes, 1) + require.Len(t, mounts, 1) + assert.Equal(t, analyticsSecret.Name, volumes[0].Secret.SecretName) + assert.Equal(t, mysqlconnection.MountPath("analytics"), mounts[0].MountPath) + assert.NotEmpty(t, firstDigest) + + itemKeys := make([]string, 0, len(volumes[0].Secret.Items)) + for _, item := range volumes[0].Secret.Items { + itemKeys = append(itemKeys, item.Key) + } + assert.ElementsMatch(t, []string{mysqlconnection.SSLCAKey}, itemKeys) + assert.NotContains(t, itemKeys, mysqlconnection.PasswordKey) + + updated := &corev1.Secret{} + require.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Namespace: analyticsSecret.Namespace, + Name: analyticsSecret.Name, + }, updated)) + updated.Data[mysqlconnection.PasswordKey] = []byte("rotated-password") + require.NoError(t, client.Update(t.Context(), updated)) + + _, _, secondDigest, err := applyMySQLBundlesToWorkload( + t.Context(), + client, + wandb, + map[string]struct{}{"analytics": {}}, + nil, + nil, + ) + require.NoError(t, err) + assert.NotEqual(t, firstDigest, secondDigest) +} + +func TestApplyMySQLBundlesToWorkloadIgnoresLegacyConnectionSecret(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + legacy := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "legacy", Namespace: "default"}, + Data: map[string][]byte{mysqlconnection.URLKey: []byte("mysql://legacy")}, + } + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(legacy).Build() + wandb := &apiv2.WeightsAndBiases{ + ObjectMeta: metav1.ObjectMeta{Name: "wandb", Namespace: "default"}, + Status: apiv2.WeightsAndBiasesStatus{MySQLStatus: map[string]apiv2.MysqlInfraStatus{ + apiv2.DefaultInstanceName: {Connection: *mysqlconnection.Connection(legacy.Name)}, + }}, + } + + volumes, mounts, digest, err := applyMySQLBundlesToWorkload( + t.Context(), + client, + wandb, + map[string]struct{}{apiv2.DefaultInstanceName: {}}, + nil, + nil, + ) + require.NoError(t, err) + assert.Empty(t, volumes) + assert.Empty(t, mounts) + assert.Empty(t, digest) +} + +func mysqlBundleTestSecret(name, password string) *corev1.Secret { + return &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + Annotations: map[string]string{ + mysqlconnection.BundleVersionAnnotation: mysqlconnection.BundleVersion, + }, + }, + Data: map[string][]byte{ + mysqlconnection.URLKey: []byte("mysql://wandb@mysql:3306/wandb"), + mysqlconnection.PasswordKey: []byte(password), + mysqlconnection.SSLCAKey: []byte("CA"), + }, + } +} diff --git a/internal/controller/reconciler/mysql_lifecycle_test.go b/internal/controller/reconciler/mysql_lifecycle_test.go new file mode 100644 index 00000000..149ed157 --- /dev/null +++ b/internal/controller/reconciler/mysql_lifecycle_test.go @@ -0,0 +1,193 @@ +package reconciler + +import ( + "testing" + + mocov1beta2 "github.com/cybozu-go/moco/api/v1beta2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apiv2 "github.com/wandb/operator/api/v2" + "github.com/wandb/operator/internal/controller/common" + "github.com/wandb/operator/internal/controller/infra/managed/mysql/moco" + 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/fake" +) + +func TestReconcileStaleManagedMySQLDetachesRemovedInstance(t *testing.T) { + scheme := mysqlLifecycleScheme(t) + wandb := mysqlLifecycleWandb(apiv2.DetachOnDelete) + labels := moco.BuildWandbMysqlLabels(wandb, "analytics") + cluster := &mocov1beta2.MySQLCluster{ObjectMeta: metav1.ObjectMeta{ + Name: "analytics", + Namespace: "database", + Labels: labels, + Annotations: map[string]string{moco.RetentionPolicyAnnotation: string(apiv2.DetachOnDelete)}, + }} + configMap := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{ + Name: moco.MyCnfConfigMapName(cluster.Name), + Namespace: cluster.Namespace, + Labels: labels, + }} + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(wandb, cluster, configMap).Build() + + require.NoError(t, reconcileStaleManagedMySQL(t.Context(), client, wandb)) + require.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Name: cluster.Name, Namespace: cluster.Namespace, + }, cluster)) + assert.Equal(t, "true", cluster.Annotations[moco.DetachedAnnotation]) + require.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Name: configMap.Name, Namespace: configMap.Namespace, + }, configMap)) + assert.Equal(t, "true", configMap.Annotations[moco.DetachedAnnotation]) +} + +func TestReconcileStaleManagedMySQLPurgesRemovedInstance(t *testing.T) { + scheme := mysqlLifecycleScheme(t) + wandb := mysqlLifecycleWandb(apiv2.PurgeOnDelete) + labels := moco.BuildWandbMysqlLabels(wandb, "analytics") + cluster := &mocov1beta2.MySQLCluster{ObjectMeta: metav1.ObjectMeta{ + Name: "analytics", + Namespace: "database", + Labels: labels, + Annotations: map[string]string{moco.RetentionPolicyAnnotation: string(apiv2.PurgeOnDelete)}, + }} + configMap := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{ + Name: moco.MyCnfConfigMapName(cluster.Name), + Namespace: cluster.Namespace, + Labels: labels, + }} + pvc := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{ + Name: "mysql-data-moco-analytics-0", Namespace: cluster.Namespace, Labels: labels, + }} + credentials := &corev1.Secret{ObjectMeta: metav1.ObjectMeta{ + Name: "moco-" + cluster.Name, Namespace: cluster.Namespace, + }} + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(wandb, cluster, configMap, pvc, credentials).Build() + + require.NoError(t, reconcileStaleManagedMySQL(t.Context(), client, wandb)) + assert.True(t, apierrors.IsNotFound(client.Get(t.Context(), types.NamespacedName{ + Name: cluster.Name, Namespace: cluster.Namespace, + }, &mocov1beta2.MySQLCluster{}))) + assert.True(t, apierrors.IsNotFound(client.Get(t.Context(), types.NamespacedName{ + Name: configMap.Name, Namespace: configMap.Namespace, + }, &corev1.ConfigMap{}))) + assert.True(t, apierrors.IsNotFound(client.Get(t.Context(), types.NamespacedName{ + Name: pvc.Name, Namespace: pvc.Namespace, + }, &corev1.PersistentVolumeClaim{}))) + assert.True(t, apierrors.IsNotFound(client.Get(t.Context(), types.NamespacedName{ + Name: credentials.Name, Namespace: credentials.Namespace, + }, &corev1.Secret{}))) +} + +func TestReconcileStaleManagedMySQLPurgeKeepsOtherInstanceStorage(t *testing.T) { + scheme := mysqlLifecycleScheme(t) + wandb := mysqlLifecycleWandb(apiv2.PurgeOnDelete) + wandb.Spec.MySQL["default"] = apiv2.MySQLSpec{ManagedMysql: &apiv2.ManagedMysqlSpec{ + Name: "primary", Namespace: "database", + }} + // A cluster from before the per-instance labels existed: only the + // deployment-wide labels and a legacy owner reference identify it. + legacyLabels := common.BuildWandbLabels(wandb, moco.MysqlModuleName) + cluster := &mocov1beta2.MySQLCluster{ObjectMeta: metav1.ObjectMeta{ + Name: "analytics", Namespace: "database", Labels: legacyLabels, + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: apiv2.GroupVersion.String(), + Kind: "WeightsAndBiases", + Name: wandb.Name, + UID: wandb.UID, + }}, + }} + stalePVC := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{ + Name: "mysql-data-moco-analytics-0", Namespace: "database", Labels: legacyLabels, + }} + otherInstancePVC := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{ + Name: "mysql-data-moco-primary-0", Namespace: "database", Labels: legacyLabels, + }} + client := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(wandb, cluster, stalePVC, otherInstancePVC).Build() + + require.NoError(t, reconcileStaleManagedMySQL(t.Context(), client, wandb)) + assert.True(t, apierrors.IsNotFound(client.Get(t.Context(), types.NamespacedName{ + Name: stalePVC.Name, Namespace: stalePVC.Namespace, + }, &corev1.PersistentVolumeClaim{})), "stale instance PVC should be purged") + assert.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Name: otherInstancePVC.Name, Namespace: otherInstancePVC.Namespace, + }, &corev1.PersistentVolumeClaim{}), "still-desired instance PVC must survive") +} + +func TestReconcileStaleManagedMySQLKeepsRenamedInstanceKey(t *testing.T) { + scheme := mysqlLifecycleScheme(t) + wandb := mysqlLifecycleWandb(apiv2.DetachOnDelete) + wandb.Spec.MySQL["renamed"] = apiv2.MySQLSpec{ManagedMysql: &apiv2.ManagedMysqlSpec{ + Name: "analytics", Namespace: "database", + }} + cluster := &mocov1beta2.MySQLCluster{ObjectMeta: metav1.ObjectMeta{ + Name: "analytics", + Namespace: "database", + Labels: moco.BuildWandbMysqlLabels(wandb, "analytics"), + }} + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(wandb, cluster).Build() + + require.NoError(t, reconcileStaleManagedMySQL(t.Context(), client, wandb)) + require.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Name: cluster.Name, Namespace: cluster.Namespace, + }, cluster)) + assert.Empty(t, cluster.Annotations[moco.DetachedAnnotation]) +} + +func TestReconcileStaleManagedMySQLReadoptsRestoredInstance(t *testing.T) { + scheme := mysqlLifecycleScheme(t) + wandb := mysqlLifecycleWandb(apiv2.DetachOnDelete) + wandb.Spec.MySQL["analytics"] = apiv2.MySQLSpec{ManagedMysql: &apiv2.ManagedMysqlSpec{ + Name: "analytics", Namespace: "database", + }} + labels := moco.BuildWandbMysqlLabels(wandb, "analytics") + cluster := &mocov1beta2.MySQLCluster{ObjectMeta: metav1.ObjectMeta{ + Name: "analytics", + Namespace: "database", + Labels: labels, + Annotations: map[string]string{moco.DetachedAnnotation: "true"}, + }} + configMap := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{ + Name: moco.MyCnfConfigMapName(cluster.Name), + Namespace: cluster.Namespace, + Labels: labels, + Annotations: map[string]string{moco.DetachedAnnotation: "true"}, + }} + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(wandb, cluster, configMap).Build() + + require.NoError(t, reconcileStaleManagedMySQL(t.Context(), client, wandb)) + require.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Name: cluster.Name, Namespace: cluster.Namespace, + }, cluster)) + assert.Empty(t, cluster.Annotations[moco.DetachedAnnotation]) + require.NoError(t, client.Get(t.Context(), types.NamespacedName{ + Name: configMap.Name, Namespace: configMap.Namespace, + }, configMap)) + assert.Empty(t, configMap.Annotations[moco.DetachedAnnotation]) +} + +func mysqlLifecycleWandb(policy apiv2.OnDeletePolicy) *apiv2.WeightsAndBiases { + return &apiv2.WeightsAndBiases{ + ObjectMeta: metav1.ObjectMeta{ + Name: "wandb", Namespace: "app", UID: types.UID("wandb-uid"), + }, + Spec: apiv2.WeightsAndBiasesSpec{ + RetentionPolicy: apiv2.RetentionPolicy{OnDelete: policy}, + MySQL: map[string]apiv2.MySQLSpec{}, + }, + } +} + +func mysqlLifecycleScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, apiv2.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, mocov1beta2.AddToScheme(scheme)) + return scheme +} diff --git a/internal/controller/reconciler/pods.go b/internal/controller/reconciler/pods.go index 643de0eb..6e7ba5c3 100644 --- a/internal/controller/reconciler/pods.go +++ b/internal/controller/reconciler/pods.go @@ -137,8 +137,19 @@ func resolveContainers(app serverManifest.Application, wandb *v2.WeightsAndBiase return containers } +type envResolution struct { + EnvVars []v1.EnvVar + MySQLInstances map[string]struct{} +} + func resolveEnvvars(ctx context.Context, client ctrlClient.Client, wandb *v2.WeightsAndBiases, manifest serverManifest.Manifest, commonEnvs []string, envs []serverManifest.EnvVar) ([]v1.EnvVar, error) { + resolved, err := resolveEnvvarsWithDependencies(ctx, client, wandb, manifest, commonEnvs, envs) + return resolved.EnvVars, err +} + +func resolveEnvvarsWithDependencies(ctx context.Context, client ctrlClient.Client, wandb *v2.WeightsAndBiases, manifest serverManifest.Manifest, commonEnvs []string, envs []serverManifest.EnvVar) (envResolution, error) { logger := logx.GetSlog(ctx) + result := envResolution{MySQLInstances: map[string]struct{}{}} var combinedEnvs []serverManifest.EnvVar for _, commonVars := range commonEnvs { if envvars, ok := manifest.CommonEnvvars[commonVars]; ok { @@ -214,6 +225,14 @@ func resolveEnvvars(ctx context.Context, client ctrlClient.Client, wandb *v2.Wei if !ok { continue } + instance := src.Name + if instance == "" { + instance = v2.DefaultInstanceName + } + if _, found := wandb.Status.MySQLStatus[instance]; !found { + instance = v2.DefaultInstanceName + } + result.MySQLInstances[instance] = struct{}{} selector := status.Connection.URL // Record for potential direct assignment case singleSecretSelector = selector @@ -382,7 +401,7 @@ func resolveEnvvars(ctx context.Context, client ctrlClient.Client, wandb *v2.Wei ctrlClient.MatchingLabels{"app.kubernetes.io/name": src.Name}, ) if err != nil { - return nil, err + return envResolution{}, err } if len(serviceList.Items) == 0 || len(serviceList.Items[0].Spec.Ports) == 0 { // Can't resolve; skip this component @@ -477,7 +496,8 @@ func resolveEnvvars(ctx context.Context, client ctrlClient.Client, wandb *v2.Wei Value: strings.Join(components, ","), }) } - return envVars, nil + result.EnvVars = envVars + return result, nil } func resolveVolumeMounts(ctx context.Context, manifest serverManifest.Manifest, commonvms []string, vms []serverManifest.VolumeMount) ([]v1.Volume, []v1.VolumeMount, error) { diff --git a/internal/controller/reconciler/pods_instance_test.go b/internal/controller/reconciler/pods_instance_test.go index 90ed00b0..54892629 100644 --- a/internal/controller/reconciler/pods_instance_test.go +++ b/internal/controller/reconciler/pods_instance_test.go @@ -78,3 +78,32 @@ func TestResolveEnvvarsMysqlMissingInstanceFallsBackToDefault(t *testing.T) { t.Fatalf("expected fallback to default-conn, got %q", got) } } + +func TestResolveEnvvarsMysqlReportsResolvedFallbackDependency(t *testing.T) { + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("failed adding corev1 to scheme: %v", err) + } + client := fake.NewClientBuilder().WithScheme(scheme).Build() + envs := []serverManifest.EnvVar{ + {Name: "MYSQL", Sources: []serverManifest.EnvSource{{Type: "mysql", Name: "missing"}}}, + } + + resolved, err := resolveEnvvarsWithDependencies( + context.Background(), + client, + wandbWithTwoMysqlInstances(), + serverManifest.Manifest{}, + nil, + envs, + ) + if err != nil { + t.Fatalf("resolveEnvvarsWithDependencies returned error: %v", err) + } + if _, ok := resolved.MySQLInstances[apiv2.DefaultInstanceName]; !ok { + t.Fatalf("expected default MySQL dependency, got %+v", resolved.MySQLInstances) + } + if _, ok := resolved.MySQLInstances["missing"]; ok { + t.Fatalf("unexpected unresolved MySQL dependency: %+v", resolved.MySQLInstances) + } +} diff --git a/internal/controller/reconciler/reconcile_v2.go b/internal/controller/reconciler/reconcile_v2.go index 2da77ebf..69c02399 100644 --- a/internal/controller/reconciler/reconcile_v2.go +++ b/internal/controller/reconciler/reconcile_v2.go @@ -307,7 +307,10 @@ func Reconcile( ///////////////////////// // Write Infra State redisConditions := redisWriteState(ctx, client, wandb, manifest) - mysqlConditions := mysqlWriteState(ctx, client, wandb, manifest) + mysqlConditions, err := mysqlWriteState(ctx, client, wandb, manifest) + if err != nil { + return ctrl.Result{}, err + } objectStoreConditions, objectStoreConnection := objectStoreWriteState(ctx, client, wandb, manifest) kafkaConditions := kafkaWriteState(ctx, client, wandb, manifest) clickHouseConditions := clickHouseWriteState(ctx, client, wandb, manifest) @@ -602,10 +605,11 @@ func reconcileApplications( app = applyWandbProbeDefaults(app, wandb.Spec.Wandb.Probes) - envVars, err := resolveEnvvars(ctx, client, wandb, manifest, app.CommonEnvs, app.Env) + envResolution, err := resolveEnvvarsWithDependencies(ctx, client, wandb, manifest, app.CommonEnvs, app.Env) if err != nil { return ctrl.Result{}, err } + envVars := envResolution.EnvVars envVars, err = injectManagedWorkloadTelemetryEnvvars(ctx, client, wandb, manifest, app, envVars, telemetryConfig) if err != nil { return ctrl.Result{}, err @@ -636,6 +640,18 @@ func reconcileApplications( if err != nil { return ctrl.Result{}, err } + var mysqlChecksum string + volumes, volumeMounts, mysqlChecksum, err = applyMySQLBundlesToWorkload( + ctx, + client, + wandb, + envResolution.MySQLInstances, + volumes, + volumeMounts, + ) + if err != nil { + return ctrl.Result{}, err + } // spec.global.proxy env: after CA (so both are present) and before legacy // overrides (so legacyOverrides can still override/blank any proxy var). @@ -670,6 +686,7 @@ func reconcileApplications( application.Spec.PodTemplate.Spec.Affinity = wandb.Spec.Affinity application.Spec.PodTemplate.Spec.Tolerations = *wandb.Spec.Tolerations setCustomCACertsChecksumAnnotation(&application.Spec.PodTemplate, caChecksum) + setMySQLBundlesChecksumAnnotation(&application.Spec.PodTemplate, mysqlChecksum) application.Spec.HpaTemplate = ResolveAutoscaling(app, wandb) @@ -1261,15 +1278,35 @@ func runMigrations(ctx context.Context, client ctrlClient.Client, wandb *apiv2.W if apiErrors.IsNotFound(err) { - envVars, err := resolveEnvvars(ctx, client, wandb, manifest, migrationTask.CommonEnvs, migrationTask.Env) + envResolution, err := resolveEnvvarsWithDependencies( + ctx, + client, + wandb, + manifest, + migrationTask.CommonEnvs, + migrationTask.Env, + ) if err != nil { return ctrl.Result{}, err } + envVars := envResolution.EnvVars volumes, volumeMounts, err := resolveVolumeMounts(ctx, manifest, migrationTask.CommonVolumeMounts, migrationTask.VolumeMounts) if err != nil { return ctrl.Result{}, err } + var mysqlChecksum string + volumes, volumeMounts, mysqlChecksum, err = applyMySQLBundlesToWorkload( + ctx, + client, + wandb, + envResolution.MySQLInstances, + volumes, + volumeMounts, + ) + if err != nil { + return ctrl.Result{}, err + } var caChecksum string envVars, volumes, volumeMounts, caChecksum, err = applyCustomCACertsToWorkload(ctx, client, wandb, envVars, volumes, volumeMounts) @@ -1302,6 +1339,7 @@ func runMigrations(ctx context.Context, client ctrlClient.Client, wandb *apiv2.W }, } setCustomCACertsChecksumAnnotation(&podTemplate, caChecksum) + setMySQLBundlesChecksumAnnotation(&podTemplate, mysqlChecksum) job = &batchv1.Job{ ObjectMeta: metav1.ObjectMeta{ @@ -1340,6 +1378,37 @@ func runMigrations(ctx context.Context, client ctrlClient.Client, wandb *apiv2.W return ctrl.Result{RequeueAfter: 5 * time.Second}, nil } + if job.Status.Succeeded == 0 { + envResolution, err := resolveEnvvarsWithDependencies( + ctx, + client, + wandb, + manifest, + migrationTask.CommonEnvs, + migrationTask.Env, + ) + if err != nil { + return ctrl.Result{}, err + } + _, _, mysqlChecksum, err := applyMySQLBundlesToWorkload( + ctx, + client, + wandb, + envResolution.MySQLInstances, + nil, + nil, + ) + if err != nil { + return ctrl.Result{}, err + } + if mysqlChecksum != "" && job.Spec.Template.Annotations[mysqlBundlesChecksumAnnotation] != mysqlChecksum { + if err := deleteJobCascading(ctx, client, job); err != nil && !apiErrors.IsNotFound(err) { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: 5 * time.Second}, nil + } + } + if job.Status.Succeeded > 0 { jobStatus.Succeeded = true jobStatus.Phase = migrationPhaseSucceeded diff --git a/internal/controller/weightsandbiases_controller.go b/internal/controller/weightsandbiases_controller.go index 965264ee..b84a69f4 100644 --- a/internal/controller/weightsandbiases_controller.go +++ b/internal/controller/weightsandbiases_controller.go @@ -19,9 +19,12 @@ package controller import ( "context" "fmt" + "sort" mocov1beta2 "github.com/cybozu-go/moco/api/v1beta2" apiv2 "github.com/wandb/operator/api/v2" + "github.com/wandb/operator/internal/controller/common" + managedmysqlmoco "github.com/wandb/operator/internal/controller/infra/managed/mysql/moco" v2 "github.com/wandb/operator/internal/controller/reconciler" "github.com/wandb/operator/pkg/utils" "github.com/wandb/operator/pkg/wandb/spec/channel/deployer" @@ -55,6 +58,8 @@ type WeightsAndBiasesReconciler struct { TelemetryConfigRef types.NamespacedName } +const mysqlSecretDependencyIndex = "mysqlSecretDependencies" + //+kubebuilder:rbac:groups="",resources=configmaps;events;persistentvolumeclaims;secrets;serviceaccounts;services,verbs=update;delete;get;list;create;patch;watch //+kubebuilder:rbac:groups="",resources=endpoints;nodes;nodes/spec;nodes/stats;nodes/metrics;nodes/proxy;namespaces;namespaces/status;replicationcontrollers;replicationcontrollers/status;resourcequotas;pods;pods/log;pods/status,verbs=get;list;watch //+kubebuilder:rbac:groups=apps,resources=deployments/status;daemonsets/status;replicasets/status;statefulsets/status,verbs=get @@ -133,6 +138,15 @@ func (r *WeightsAndBiasesReconciler) Delete(e event.DeleteEvent) bool { // SetupWithManager sets up the controller with the Manager. func (r *WeightsAndBiasesReconciler) SetupWithManager(mgr ctrl.Manager) error { + if err := mgr.GetFieldIndexer().IndexField( + context.Background(), + &apiv2.WeightsAndBiases{}, + mysqlSecretDependencyIndex, + mysqlSecretDependencies, + ); err != nil { + return err + } + var b = ctrl.NewControllerManagedBy(mgr). For(&apiv2.WeightsAndBiases{}). // Applications carry plain (non-controller) owner refs; without @@ -141,13 +155,14 @@ func (r *WeightsAndBiasesReconciler) SetupWithManager(mgr ctrl.Manager) error { Owns(&apiv2.Application{}, builder.MatchEveryOwner). Owns(&batchv1.Job{}). Owns(&corev1.Secret{}). + Watches(&corev1.Secret{}, handler.EnqueueRequestsFromMapFunc(r.mapMySQLSecretToWandb)). Owns(&corev1.ConfigMap{}). Owns(&networkingv1.Ingress{}) if utils.IsRegistered(r.Scheme, &gatewayv1.Gateway{}) { b = b.Watches(&gatewayv1.Gateway{}, handler.EnqueueRequestsFromMapFunc(r.mapGatewayToWandb)) } if utils.IsRegistered(r.Scheme, &mocov1beta2.MySQLCluster{}) { - b = b.Owns(&mocov1beta2.MySQLCluster{}) + b = b.Watches(&mocov1beta2.MySQLCluster{}, handler.EnqueueRequestsFromMapFunc(mapMocoToWandb)) } if r.TelemetryConfigRef.Name != "" { b = b.Watches( @@ -160,6 +175,79 @@ func (r *WeightsAndBiasesReconciler) SetupWithManager(mgr ctrl.Manager) error { return b.Complete(r) } +func mapMocoToWandb(_ context.Context, obj client.Object) []ctrl.Request { + name := obj.GetLabels()[common.WandbNameLabel] + namespace := obj.GetLabels()[common.WandbNamespaceLabel] + if name == "" || namespace == "" { + return nil + } + return []ctrl.Request{{NamespacedName: types.NamespacedName{Name: name, Namespace: namespace}}} +} + +func mysqlSecretDependencies(obj client.Object) []string { + wandb, ok := obj.(*apiv2.WeightsAndBiases) + if !ok { + return nil + } + + dependencies := map[string]struct{}{} + add := func(namespace string, selector corev1.SecretKeySelector) { + if selector.Name == "" { + return + } + dependencies[types.NamespacedName{Namespace: namespace, Name: selector.Name}.String()] = struct{}{} + } + for key, spec := range wandb.Spec.MySQL { + if external := spec.ExternalMysql; external != nil { + add(wandb.Namespace, external.Host) + add(wandb.Namespace, external.Port) + add(wandb.Namespace, external.Database) + add(wandb.Namespace, external.Username) + add(wandb.Namespace, external.Password) + add(wandb.Namespace, external.Tls) + add(wandb.Namespace, external.SslCa) + add(wandb.Namespace, external.SslCert) + add(wandb.Namespace, external.SslKey) + } + if managed := spec.ManagedMysql; managed != nil { + name := managed.Name + if name == "" { + name = managedmysqlmoco.DefaultSpecName(wandb.Name, key) + } + namespace := managed.Namespace + if namespace == "" { + namespace = wandb.Namespace + } + dependencies[types.NamespacedName{ + Namespace: namespace, + Name: "moco-" + name, + }.String()] = struct{}{} + } + } + + result := make([]string, 0, len(dependencies)) + for dependency := range dependencies { + result = append(result, dependency) + } + sort.Strings(result) + return result +} + +func (r *WeightsAndBiasesReconciler) mapMySQLSecretToWandb(ctx context.Context, obj client.Object) []ctrl.Request { + wandbList := &apiv2.WeightsAndBiasesList{} + if err := r.List(ctx, wandbList, client.MatchingFields{ + mysqlSecretDependencyIndex: client.ObjectKeyFromObject(obj).String(), + }); err != nil { + return nil + } + + requests := make([]ctrl.Request, 0, len(wandbList.Items)) + for i := range wandbList.Items { + requests = append(requests, ctrl.Request{NamespacedName: client.ObjectKeyFromObject(&wandbList.Items[i])}) + } + return requests +} + func (r *WeightsAndBiasesReconciler) loadTelemetryConfig(ctx context.Context) (v2.TelemetryRuntimeConfig, error) { if r.TelemetryConfigRef.Name == "" { return v2.DefaultTelemetryRuntimeConfig(), nil diff --git a/internal/controller/weightsandbiases_controller_test.go b/internal/controller/weightsandbiases_controller_test.go index 93c2a457..8e7509b9 100644 --- a/internal/controller/weightsandbiases_controller_test.go +++ b/internal/controller/weightsandbiases_controller_test.go @@ -190,19 +190,6 @@ var _ = Describe("WeightsAndBiases Controller V2", func() { wandb.Status.ClickHouseStatus = map[string]apiv2.ClickHouseInfraStatus{apiv2.DefaultInstanceName: {WBInfraStatus: apiv2.WBInfraStatus{Ready: true}}} Expect(k8sClient.Status().Update(ctx, wandb)).Should(Succeed()) - By("Creating the db-password secret") - secret := &v1.Secret{ - ObjectMeta: metav1.ObjectMeta{ - Name: wandbName + "-db-password", - Namespace: WandbNamespace, - }, - Data: map[string][]byte{ - "rootPassword": []byte("root-pass"), - "password": []byte("user-pass"), - }, - } - Expect(k8sClient.Create(ctx, secret)).Should(Succeed()) - wandbLookupKey := types.NamespacedName{Name: wandb.Name, Namespace: wandb.Namespace} Expect(k8sClient.Get(ctx, wandbLookupKey, wandb)).Should(Succeed()) @@ -218,6 +205,11 @@ var _ = Describe("WeightsAndBiases Controller V2", func() { NamespacedName: wandbLookupKey, }) Expect(err).Should(Succeed()) + unusedPasswordSecret := &v1.Secret{} + Expect(errors.IsNotFound(k8sClient.Get(ctx, types.NamespacedName{ + Name: wandbName + "-db-password", + Namespace: WandbNamespace, + }, unusedPasswordSecret))).To(BeTrue()) By("Setting infrastructure status to ready") Expect(k8sClient.Get(ctx, wandbLookupKey, wandb)).Should(Succeed()) @@ -251,6 +243,13 @@ var _ = Describe("WeightsAndBiases Controller V2", func() { }, timeout, interval).Should(Succeed()) Expect(job.Spec.Template.Spec.Containers[0].Name).To(Equal("moco-init")) + Expect(job.Spec.Template.Spec.Containers[0].Command).To(Equal([]string{"mysql"})) + Expect(job.Spec.Template.Spec.Containers[0].Args).To(ContainElement( + "--execute=CREATE DATABASE IF NOT EXISTS `wandb_local`;", + )) + Expect(job.Spec.Template.Spec.Containers[0].Env).To(ContainElement( + HaveField("Name", "MYSQL_PWD"), + )) }) It("Should create application components when infrastructure is ready", func() { diff --git a/internal/controller/weightsandbiases_mysql_watch_test.go b/internal/controller/weightsandbiases_mysql_watch_test.go new file mode 100644 index 00000000..c2a0891b --- /dev/null +++ b/internal/controller/weightsandbiases_mysql_watch_test.go @@ -0,0 +1,115 @@ +package controller + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apiv2 "github.com/wandb/operator/api/v2" + "github.com/wandb/operator/internal/controller/common" + corev1 "k8s.io/api/core/v1" + 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/fake" +) + +func TestMySQLSecretDependencies(t *testing.T) { + selector := func(name, key string) corev1.SecretKeySelector { + return corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: name}, + Key: key, + } + } + wandb := &apiv2.WeightsAndBiases{ + ObjectMeta: metav1.ObjectMeta{Name: "wandb", Namespace: "app"}, + Spec: apiv2.WeightsAndBiasesSpec{MySQL: map[string]apiv2.MySQLSpec{ + apiv2.DefaultInstanceName: {ExternalMysql: &apiv2.MysqlConnection{ + Host: selector("mysql-address", "host"), + Port: selector("mysql-address", "port"), + Database: selector("mysql-credentials", "database"), + Username: selector("mysql-credentials", "username"), + Password: selector("mysql-credentials", "password"), + SslCa: selector("mysql-tls", "ca"), + }}, + "analytics": {ManagedMysql: &apiv2.ManagedMysqlSpec{ + Name: "analytics", + Namespace: "database", + }}, + }}, + } + + assert.Equal(t, []string{ + "app/mysql-address", + "app/mysql-credentials", + "app/mysql-tls", + "database/moco-analytics", + }, mysqlSecretDependencies(wandb)) +} + +func TestMySQLSecretDependenciesDerivesManagedDefaults(t *testing.T) { + wandb := &apiv2.WeightsAndBiases{ + ObjectMeta: metav1.ObjectMeta{Name: "wandb", Namespace: "app"}, + Spec: apiv2.WeightsAndBiasesSpec{MySQL: map[string]apiv2.MySQLSpec{ + apiv2.DefaultInstanceName: {ManagedMysql: &apiv2.ManagedMysqlSpec{}}, + }}, + } + + assert.Equal(t, []string{"app/moco-wandb-mysql"}, mysqlSecretDependencies(wandb)) +} + +func TestMapMySQLSecretToEveryDependentWandb(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, apiv2.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + selector := corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: "shared-mysql"}, Key: "value", + } + dependent := func(name string) *apiv2.WeightsAndBiases { + return &apiv2.WeightsAndBiases{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "app"}, + Spec: apiv2.WeightsAndBiasesSpec{MySQL: map[string]apiv2.MySQLSpec{ + apiv2.DefaultInstanceName: {ExternalMysql: &apiv2.MysqlConnection{ + Host: selector, Port: selector, Database: selector, Username: selector, Password: selector, + }}, + }}, + } + } + first := dependent("first") + second := dependent("second") + unrelated := dependent("unrelated") + unrelated.Spec.MySQL[apiv2.DefaultInstanceName].ExternalMysql.Host.Name = "another-secret" + unrelated.Spec.MySQL[apiv2.DefaultInstanceName].ExternalMysql.Port.Name = "another-secret" + unrelated.Spec.MySQL[apiv2.DefaultInstanceName].ExternalMysql.Database.Name = "another-secret" + unrelated.Spec.MySQL[apiv2.DefaultInstanceName].ExternalMysql.Username.Name = "another-secret" + unrelated.Spec.MySQL[apiv2.DefaultInstanceName].ExternalMysql.Password.Name = "another-secret" + + client := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(first, second, unrelated). + WithIndex(&apiv2.WeightsAndBiases{}, mysqlSecretDependencyIndex, mysqlSecretDependencies). + Build() + reconciler := &WeightsAndBiasesReconciler{Client: client} + requests := reconciler.mapMySQLSecretToWandb(t.Context(), &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "shared-mysql", Namespace: "app"}, + }) + + requestNames := make([]string, 0, len(requests)) + for _, request := range requests { + requestNames = append(requestNames, request.Name) + } + assert.ElementsMatch(t, []string{"first", "second"}, requestNames) +} + +func TestMapMocoToWandbUsesProvenanceLabels(t *testing.T) { + cluster := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{ + common.WandbNameLabel: "wandb", + common.WandbNamespaceLabel: "app", + }}} + + requests := mapMocoToWandb(context.Background(), cluster) + if assert.Len(t, requests, 1) { + assert.Equal(t, types.NamespacedName{Name: "wandb", Namespace: "app"}, requests[0].NamespacedName) + } +} diff --git a/internal/webhook/v2/weightsandbiases_webhook.go b/internal/webhook/v2/weightsandbiases_webhook.go index a361020c..299bddd8 100644 --- a/internal/webhook/v2/weightsandbiases_webhook.go +++ b/internal/webhook/v2/weightsandbiases_webhook.go @@ -110,7 +110,7 @@ func (d *WeightsAndBiasesCustomDefaulter) Default(ctx context.Context, obj runti wandb.Spec.Wandb.InternalServiceAuth.Enabled = ptr.To(true) } - if wandb.Spec.Wandb.InternalServiceAuth.OIDCIssuer == "" && wandb.Spec.Wandb.InternalServiceAuth.Enabled != nil && *wandb.Spec.Wandb.InternalServiceAuth.Enabled{ + if wandb.Spec.Wandb.InternalServiceAuth.OIDCIssuer == "" && wandb.Spec.Wandb.InternalServiceAuth.Enabled != nil && *wandb.Spec.Wandb.InternalServiceAuth.Enabled { wandb.Spec.Wandb.InternalServiceAuth.OIDCIssuer = "https://kubernetes.default.svc.cluster.local" } @@ -454,6 +454,8 @@ func validateWandbSpec(wandb *appsv2.WeightsAndBiases) field.ErrorList { func validateMySQLSpec(wandb *appsv2.WeightsAndBiases) field.ErrorList { var errors field.ErrorList mysqlPath := field.NewPath("spec").Child("mysql") + _, hasPendingLegacyMySQL := wandb.Annotations[v1.MySQLPendingAnnotation] + managedLocations := map[string]string{} errors = append(errors, validateHasDefaultInstance(wandb.Spec.MySQL, mysqlPath)...) @@ -467,6 +469,18 @@ func validateMySQLSpec(wandb *appsv2.WeightsAndBiases) field.ErrorList { )) } if managed := spec.ManagedMysql; managed != nil { + if managed.Name != "" { + location := managed.Namespace + "/" + managed.Name + if existingInstance, found := managedLocations[location]; found { + errors = append(errors, field.Invalid( + instancePath.Child("managedMysql").Child("name"), + managed.Name, + fmt.Sprintf("managed MySQL resource is already used by instance %q", existingInstance), + )) + } else { + managedLocations[location] = key + } + } if managed.Replicas != 0 && !appsv2.ValidMysqlReplicaCount(managed.Replicas) { errors = append(errors, field.Invalid( instancePath.Child("managedMysql").Child("replicas"), @@ -475,11 +489,40 @@ func validateMySQLSpec(wandb *appsv2.WeightsAndBiases) field.ErrorList { )) } } + if external := spec.ExternalMysql; external != nil && !hasPendingLegacyMySQL { + externalPath := instancePath.Child("externalMysql") + errors = append(errors, validateRequiredSecretSelector(external.Host, externalPath.Child("host"))...) + errors = append(errors, validateRequiredSecretSelector(external.Port, externalPath.Child("port"))...) + errors = append(errors, validateRequiredSecretSelector(external.Database, externalPath.Child("database"))...) + errors = append(errors, validateRequiredSecretSelector(external.Username, externalPath.Child("username"))...) + errors = append(errors, validateRequiredSecretSelector(external.Password, externalPath.Child("password"))...) + errors = append(errors, validateOptionalSecretSelector(external.Tls, externalPath.Child("tls"))...) + errors = append(errors, validateOptionalSecretSelector(external.SslCa, externalPath.Child("sslCa"))...) + errors = append(errors, validateOptionalSecretSelector(external.SslCert, externalPath.Child("sslCert"))...) + errors = append(errors, validateOptionalSecretSelector(external.SslKey, externalPath.Child("sslKey"))...) + + hasCert := external.SslCert.Name != "" || external.SslCert.Key != "" + hasKey := external.SslKey.Name != "" || external.SslKey.Key != "" + if hasCert != hasKey { + errors = append(errors, field.Invalid( + externalPath, + "", + "sslCert and sslKey must be configured together", + )) + } + } } return errors } +func validateOptionalSecretSelector(selector corev1.SecretKeySelector, path *field.Path) field.ErrorList { + if selector.Name == "" && selector.Key == "" { + return nil + } + return validateRequiredSecretSelector(selector, path) +} + func validateRedisSpec(wandb *appsv2.WeightsAndBiases) field.ErrorList { var errors field.ErrorList redisPath := field.NewPath("spec").Child("redis")