Merge pull request #62 from intelops/vaultHA

Added feature to support vault HA

charts/vault-cred/Chart.yaml

@@ -15,7 +15,7 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.1.2
+version: 0.1.5
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to

charts/vault-cred/templates/deployment.yaml

@@ -1,3 +1,4 @@
+
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -42,6 +43,10 @@ spec:
               value: "{{ .Values.env.logLevel }}"
             - name: VAULT_ADDR
               value: "{{ .Values.vault.vaultAddress }}"
+            - name: VAULT_NODE_ADDRESSES
+              value: "{{ .Values.vault.vaultNodeAddresses }}"
+            - name: HA_ENABLED
+              value: "{{ .Values.vault.haEnabled }}"
             - name: VAULT_SECRET_NAME
               value: "{{ .Values.vault.secretName }}"            
             - name: VAULT_SECRET_TOKEN_KEY_NAME

charts/vault-cred/values.yaml

@@ -44,7 +44,10 @@ env:
   logLevel: info
 
 vault:
-  vaultAddress: http://vault:8200
+  haEnabled: true
+  vaultAddress: http://vault-hash:8200
+  vaultLeaderAddress: vault-hash-0.vault-hash-internal:8200
+  vaultNodeAddresses: "http://vault-hash-0:8200,http://vault-hash-1:8200,http://vault-hash-2:8200"
   secretName: vault-server
   secretTokenKeyName: roottoken
   secretUnSealKeyPrefix: unsealkey

config/config.go

@@ -15,7 +15,9 @@ type Configuration struct {
 }
 
 type VaultEnv struct {
+	HAEnabled                  bool          `envconfig:"HA_ENABLED" default:"true"`
 	Address                    string        `envconfig:"VAULT_ADDR" required:"true"`
+	NodeAddresses              []string      `envconfig:"VAULT_NODE_ADDRESSES" required:"true"`
 	CACert                     string        `envconfig:"VAULT_CACERT" required:"false"`
 	ReadTimeout                time.Duration `envconfig:"VAULT_READ_TIMEOUT" default:"60s"`
 	MaxRetries                 int           `envconfig:"VAULT_MAX_RETRIES" default:"5"`

internal/client/vault.go

@@ -3,6 +3,7 @@ package client
 import (
 	"context"
 	"encoding/base64"
+	"fmt"
 
 	"github.com/hashicorp/go-retryablehttp"
 	"github.com/hashicorp/vault/api"
@@ -72,7 +73,7 @@ func NewVaultClientForVaultToken(log logging.Logger, conf config.VaultEnv) (*Vau
 func NewVaultClient(log logging.Logger, conf config.VaultEnv) (*VaultClient, error) {
 	cfg, err := prepareVaultConfig(conf)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("error in vault config, %v", err)
 	}
 
 	c, err := api.NewClient(cfg)
@@ -175,3 +176,17 @@ func (vc *VaultClient) DeleteCredential(ctx context.Context, mountPath, secretPa
 	}
 	return
 }
+
+func (vc *VaultClient) JoinRaftCluster(leaderAddress string) error {
+	req := &api.RaftJoinRequest{
+		Retry:         true,
+		LeaderAPIAddr: leaderAddress,
+	}
+
+	res, err := vc.c.Sys().RaftJoin(req)
+	if err != nil {
+		return fmt.Errorf("failed to join the Raft cluster: %v", err)
+	}
+	vc.log.Debug("Raft Joined status", res.Joined)
+	return nil
+}

internal/client/vault_seal.go

@@ -18,6 +18,7 @@ func (vc *VaultClient) IsVaultSealed() (bool, error) {
 }
 
 func (vc *VaultClient) Unseal() error {
+
 	status, err := vc.c.Sys().SealStatus()
 	if err != nil {
 		return err
@@ -51,7 +52,9 @@ func (vc *VaultClient) Unseal() error {
 }
 
 func (vc *VaultClient) initializeVaultSecret() error {
+
 	unsealKeys, rootToken, err := vc.generateUnsealKeys()
+
 	if err != nil {
 		return errors.WithMessage(err, "error while generating unseal keys")
 	}
@@ -92,6 +95,14 @@ func (vc *VaultClient) generateUnsealKeys() ([]string, string, error) {
 	return unsealKeys, rootToken, err
 }
 
+func (vc *VaultClient) Leader() (string, error) {
+	res, err := vc.c.Sys().Leader()
+	if err != nil {
+		return "", err
+	}
+	return res.LeaderAddress, nil
+}
+
 func (vc *VaultClient) getVaultSecretValues() (string, []string, error) {
 	k8s, err := NewK8SClient(vc.log)
 	if err != nil {

internal/job/vault_seal_watcher.go

@@ -1,11 +1,17 @@
 package job
 
 import (
+	"fmt"
+
 	"github.com/intelops/go-common/logging"
 	"github.com/intelops/vault-cred/config"
 	"github.com/intelops/vault-cred/internal/client"
 )
 
+// this var is set to true after leader created for the very first time
+// remove this later, after learning the correct usecase of Leader() api
+var leaderCreated bool
+
 type VaultSealWatcher struct {
 	log       logging.Logger
 	conf      config.VaultEnv
@@ -29,36 +35,116 @@ func (v *VaultSealWatcher) CronSpec() string {
 }
 
 func (v *VaultSealWatcher) Run() {
-	v.log.Debug("started vault seal watcher job")
+	v.log.Debugf("started vault seal watcher job with vault HA: %v", v.conf.HAEnabled)
+
+	if v.conf.HAEnabled {
+		if len(v.conf.NodeAddresses) != 3 {
+			v.log.Errorf("vault HA node count %d is not valid", len(v.conf.NodeAddresses))
+			return
+		}
+
+		if err := v.handleUnsealForHAVault(); err != nil {
+			v.log.Errorf("%s", err)
+		}
+	} else {
+		if err := v.handleUnsealForNonHAVault(); err != nil {
+			v.log.Errorf("%s", err)
+		}
+	}
+}
+
+func (v *VaultSealWatcher) handleUnsealForNonHAVault() error {
 	vc, err := client.NewVaultClient(v.log, v.conf)
 	if err != nil {
-		v.log.Errorf("%s", err)
-		return
+		return err
 	}
 
 	res, err := vc.IsVaultSealed()
 	if err != nil {
-		v.log.Errorf("failed to get vault seal status, %s", err)
-		return
+		return fmt.Errorf("failed to get vault seal status, %s", err)
 	}
 
 	if res {
 		v.log.Info("vault is sealed, trying to unseal")
 		err := vc.Unseal()
 		if err != nil {
-			v.log.Errorf("failed to unseal vault, %s", err)
-			return
+			return fmt.Errorf("failed to unseal vault, %s", err)
 		}
 		v.log.Info("vault unsealed executed")
 
 		res, err := vc.IsVaultSealed()
 		if err != nil {
-			v.log.Errorf("failed to get vault seal status, %s", err)
-			return
+			return fmt.Errorf("failed to get vault seal status, %s", err)
 		}
 		v.log.Infof("vault sealed status: %v", res)
-		return
 	} else {
 		v.log.Debug("vault is in unsealed status")
 	}
+	return nil
+}
+
+func (v *VaultSealWatcher) handleUnsealForHAVault() error {
+	var vaultClients []*client.VaultClient
+	for _, nodeAddress := range v.conf.NodeAddresses {
+		conf := v.conf
+		conf.Address = nodeAddress
+		vc, err := client.NewVaultClient(v.log, conf)
+		if err != nil {
+			return err
+		}
+		vaultClients = append(vaultClients, vc)
+	}
+
+	sealed, err := v.isAnyNodeSealed(vaultClients)
+	if err != nil {
+		return err
+	}
+
+	if !sealed {
+		v.log.Debug("All nodes are unsealed")
+		return nil
+	}
+
+	var leaderNode string
+	for _, vc := range vaultClients {
+		if leader, err := vc.Leader(); err == nil && leader != "" {
+			leaderNode = leader
+			break
+		}
+	}
+	v.log.Infof("Found leader node: %v", leaderNode)
+
+	for index, vc := range vaultClients {
+		if len(leaderNode) > 0 {
+			err := vc.JoinRaftCluster(leaderNode)
+			if err != nil {
+				return fmt.Errorf("failed to join the HA cluster by node index: %v, error: %v", index, err)
+			}
+			v.log.Info("Node %s joined leader %s", v.conf.Address, leaderNode)
+		}
+
+		if err := vc.Unseal(); err != nil {
+			return fmt.Errorf("failed to unseal vault on node %s, %v", v.conf.Address, err)
+		}
+
+		v.log.Info("Node %s successfully Unsealed", v.conf.Address)
+	}
+	return nil
+}
+
+func (v *VaultSealWatcher) isAnyNodeSealed(vaultClients []*client.VaultClient) (bool, error) {
+	sealedStatus := false
+	for _, vc := range vaultClients {
+		res, err := vc.IsVaultSealed()
+		if err != nil {
+			return false, fmt.Errorf("failed to get vault seal status for %s, %v", v.conf.Address, err)
+		}
+
+		if res {
+			sealedStatus = true
+		}
+
+		v.log.Debugf("vault node %s seal status %s", v.conf.Address, res)
+	}
+	return sealedStatus, nil
 }