Merge pull request #1121 from mumoshu/deploy-locking-lib

Add deploy locking library

deploy/configmap.go

@@ -0,0 +1,128 @@
+package deploy
+
+import (
+	"context"
+	"encoding/json"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type ConfigMapValue struct {
+	Locked  bool              `json:"locked"`
+	History []LockHistoryItem `json:"history"`
+}
+
+type LockHistoryItem struct {
+	User   string      `json:"user"`
+	Action LockAction  `json:"action"`
+	At     metav1.Time `json:"at"`
+	Reason string      `json:"reason"`
+}
+
+type LockAction string
+
+const (
+	LockActionLock   LockAction = "lock"
+	LockActionUnlock LockAction = "unlock"
+
+	MaxHistoryItems = 3
+)
+
+func configMapValueToStr(value ConfigMapValue) (string, error) {
+	data, err := json.Marshal(value)
+	if err != nil {
+		return "", err
+	}
+
+	return string(data), nil
+}
+
+func strToConfigMapValue(data string) (ConfigMapValue, error) {
+	if data == "" {
+		return ConfigMapValue{}, nil
+	}
+
+	var value ConfigMapValue
+	err := json.Unmarshal([]byte(data), &value)
+	if err != nil {
+		return ConfigMapValue{}, err
+	}
+
+	return value, nil
+}
+
+// configMapKey is a helper function that returns the key within the ConfigMap for the project and the environment
+// which is either locked or unlocked.
+func (c *Coordinator) configMapKey(project, environment string) string {
+	return project + "-" + environment
+}
+
+func (c *Coordinator) getOrCreateConfigMap(ctx context.Context) (*corev1.ConfigMap, error) {
+	configMap, err := c.getConfigMap(ctx)
+	if err != nil {
+		configMap, err = c.createConfigMap(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return configMap, nil
+}
+
+// getConfigMap creates a Kubernetes API client, and use it to retrieve the ConfigMap.
+func (c *Coordinator) getConfigMap(ctx context.Context) (*corev1.ConfigMap, error) {
+	clientset, err := c.kubernetesClientSet()
+	if err != nil {
+		return nil, err
+	}
+
+	configMap, err := clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, c.ConfigMapName, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	if configMap.Data == nil {
+		configMap.Data = make(map[string]string)
+	}
+
+	return configMap, nil
+}
+
+func (c *Coordinator) createConfigMap(ctx context.Context) (*corev1.ConfigMap, error) {
+	clientset, err := c.kubernetesClientSet()
+	if err != nil {
+		return nil, err
+	}
+
+	configMap := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: c.ConfigMapName,
+		},
+	}
+
+	configMap, err = clientset.CoreV1().ConfigMaps(c.Namespace).Create(ctx, configMap, metav1.CreateOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	if configMap.Data == nil {
+		configMap.Data = make(map[string]string)
+	}
+
+	return configMap, nil
+}
+
+func (c *Coordinator) updateConfigMap(ctx context.Context, configMap *corev1.ConfigMap) (*corev1.ConfigMap, error) {
+	clientset, err := c.kubernetesClientSet()
+	if err != nil {
+		return nil, err
+	}
+
+	configMap, err = clientset.CoreV1().ConfigMaps(c.Namespace).Update(ctx, configMap, metav1.UpdateOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	return configMap, nil
+}

deploy/kubernetes.go

@@ -0,0 +1,53 @@
+package deploy
+
+import (
+	"os"
+
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+// kubeconfigPath returns the path to the KUBECONFIG file,
+// which is either specified by the KUBECONFIG environment variable,
+// or the default path ~/.kube/config.
+func (c *Coordinator) kubeconfigPath() string {
+	kubeconfig := clientcmd.NewDefaultClientConfigLoadingRules().GetDefaultFilename()
+	if path := os.Getenv("KUBECONFIG"); path != "" {
+		kubeconfig = path
+	}
+
+	return kubeconfig
+}
+
+// kubernetesClientSet creates a Kubernetes API client
+// that uses either the KUBECONFIG file if exists, or the in-cluster configuration.
+func (c *Coordinator) kubernetesClientSet() (clientset.Interface, error) {
+	if c.clientset != nil {
+		return c.clientset, nil
+	}
+
+	var restConfig *rest.Config
+
+	kubeconfig := c.kubeconfigPath()
+	if _, err := os.Stat(kubeconfig); os.IsNotExist(err) {
+		restConfig, err = rest.InClusterConfig()
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		restConfig, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	clientset, err := clientset.NewForConfig(restConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	c.clientset = clientset
+
+	return clientset, nil
+}

deploy/lock.go

@@ -0,0 +1,199 @@
+package deploy
+
+import (
+	"context"
+	"fmt"
+
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	clientset "k8s.io/client-go/kubernetes"
+)
+
+// Coordinaator provides a way to lock and unlock deployments made via gocat.
+// The lock is used to prevent multiple deployments from happening at the same time.
+//
+// The lock information is stored in a Kubernetes ConfigMap managed by the Coordinaator.
+// The ConfigMap is created in the same namespace as gocat.
+//
+// The lock can be acquired by calling the Lock method. The lock is released by calling the Unlock method.
+// Each method takes a user ID as an argument. The user ID is used to identify the user who acquired the lock,
+// and also to verify that the user releasing the lock is the same user who acquired it, or has the necessary
+// permissions to release it.
+//
+// The methods also take a project name and an environment name as arguments. These are used to identify the
+// deployment that the lock is associated with.
+type Coordinator struct {
+	// Namespace is the namespace in which the ConfigMap is created.
+	Namespace string
+
+	// ConfigMapName is the name of the ConfigMap.
+	ConfigMapName string
+
+	clientset clientset.Interface
+}
+
+func NewCoordinator(ns, configMap string) *Coordinator {
+	return &Coordinator{
+		Namespace:     ns,
+		ConfigMapName: configMap,
+	}
+}
+
+var ErrLocked = fmt.Errorf("deployment is locked")
+var ErrAlreadyUnlocked = fmt.Errorf("deployment is already unlocked")
+
+const (
+	MaxConfigMapUpdateRetries = 3
+)
+
+// Lock acquires a lock for the given project and environment.
+//
+// Under the hood, this retries to update the ConfigMap if the update fails due to a conflict.
+func (c *Coordinator) Lock(ctx context.Context, project, environment, user, reason string) error {
+	var retried int
+	for {
+		err := c.lock(ctx, project, environment, user, reason)
+		if err == nil {
+			return nil
+		}
+
+		if kerrors.IsConflict(err) {
+			if retried >= MaxConfigMapUpdateRetries {
+				return fmt.Errorf("unable to acquire lock after %d retries: %w", MaxConfigMapUpdateRetries, err)
+			}
+
+			retried++
+			continue
+		} else {
+			return err
+		}
+	}
+}
+
+func (c *Coordinator) lock(ctx context.Context, project, environment, user, reason string) error {
+	configMap, err := c.getOrCreateConfigMap(ctx)
+	if err != nil {
+		return fmt.Errorf("unable to get or create configmap: %w", err)
+	}
+
+	key := c.configMapKey(project, environment)
+	value, err := strToConfigMapValue(configMap.Data[key])
+	if err != nil {
+		return fmt.Errorf("unable to unmarshal str into value: %w", err)
+	}
+
+	if value.Locked {
+		return ErrLocked
+	}
+
+	if n := len(value.History); n >= MaxHistoryItems {
+		value.History = value.History[n-MaxHistoryItems+1:]
+	}
+
+	value.History = append(value.History, LockHistoryItem{
+		User:   user,
+		Action: LockActionLock,
+		At:     metav1.Now(),
+		Reason: reason,
+	})
+
+	value.Locked = true
+
+	configMap.Data[key], err = configMapValueToStr(value)
+	if err != nil {
+		return err
+	}
+
+	_, err = c.updateConfigMap(ctx, configMap)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Unlock releases the lock for the given project and environment.
+//
+// The lock can be released by the same user who acquired it, or by anyone if the force argument is true.
+//
+// Under the hood, this retries to update the ConfigMap if the update fails due to a conflict.
+func (c *Coordinator) Unlock(ctx context.Context, project, environment, user string, force bool) error {
+	var retried int
+	for {
+		err := c.unlock(ctx, project, environment, user, force)
+		if err == nil {
+			return nil
+		}
+
+		if kerrors.IsConflict(err) {
+			if retried >= MaxConfigMapUpdateRetries {
+				return fmt.Errorf("unable to release lock after %d retries: %w", MaxConfigMapUpdateRetries, err)
+			}
+
+			retried++
+			continue
+		} else {
+			return err
+		}
+	}
+}
+
+func (c *Coordinator) unlock(ctx context.Context, project, environment, user string, force bool) error {
+	configMap, err := c.getOrCreateConfigMap(ctx)
+	if err != nil {
+		return err
+	}
+
+	key := c.configMapKey(project, environment)
+	value, err := strToConfigMapValue(configMap.Data[key])
+	if err != nil {
+		return err
+	}
+
+	if !value.Locked {
+		return ErrAlreadyUnlocked
+	}
+
+	if force {
+		value.Locked = false
+	} else {
+		if len(value.History) == 0 || value.History[len(value.History)-1].User != user {
+			return newNotAllowedToUnlockError(user)
+		}
+
+		if n := len(value.History); n >= MaxHistoryItems {
+			value.History = value.History[n-MaxHistoryItems+1:]
+		}
+
+		value.Locked = false
+		value.History = append(value.History, LockHistoryItem{
+			User:   user,
+			Action: LockActionUnlock,
+			At:     metav1.Now(),
+		})
+	}
+
+	configMap.Data[key], err = configMapValueToStr(value)
+	if err != nil {
+		return err
+	}
+
+	_, err = c.updateConfigMap(ctx, configMap)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+type NotAllowedTounlockError struct {
+	User string
+}
+
+func (e NotAllowedTounlockError) Error() string {
+	return fmt.Sprintf("user %s is not allowed to unlock", e.User)
+}
+
+func newNotAllowedToUnlockError(user string) NotAllowedTounlockError {
+	return NotAllowedTounlockError{User: user}
+}