Track persistent volume in k8s environement

src/go/k8s/cmd/configurator/main.go

@@ -11,23 +11,37 @@ package main
 
 import (
 	"context"
+	"crypto/tls"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"log"
+	"math/rand"
+	"net/http"
 	"os"
 	"path"
+	"path/filepath"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/hashicorp/go-multierror"
+	redpandav1alpha1 "github.com/redpanda-data/redpanda/src/go/k8s/apis/redpanda/v1alpha1"
+	labels "github.com/redpanda-data/redpanda/src/go/k8s/pkg/labels"
 	"github.com/redpanda-data/redpanda/src/go/k8s/pkg/networking"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
 	"github.com/spf13/afero"
 	"gopkg.in/yaml.v3"
+	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8sLabels "k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	k8sConfig "sigs.k8s.io/controller-runtime/pkg/client/config"
 )
 
 const (
@@ -42,6 +56,9 @@ const (
 	externalConnectivityAddressTypeEnvVar = "EXTERNAL_CONNECTIVITY_ADDRESS_TYPE"
 	hostPortEnvVar                        = "HOST_PORT"
 	proxyHostPortEnvVar                   = "PROXY_HOST_PORT"
+	dataDirPath                           = "DATA_DIR_PATH"
+
+	redpandaIDs = "redoanda-ids"
 )
 
 type brokerID int
@@ -58,6 +75,7 @@ type configuratorConfig struct {
 	redpandaRPCPort                 int
 	hostPort                        int
 	proxyHostPort                   int
+	dataDirPath                     string
 }
 
 func (c *configuratorConfig) String() string {
@@ -72,7 +90,8 @@ func (c *configuratorConfig) String() string {
 		"externalConnectivityAddressType: %s\n"+
 		"redpandaRPCPort: %d\n"+
 		"hostPort: %d\n"+
-		"proxyHostPort: %d\n",
+		"proxyHostPort: %d\n"+
+		"dataDirPath: %s\n",
 		c.hostName,
 		c.svcFQDN,
 		c.configSourceDir,
@@ -83,10 +102,11 @@ func (c *configuratorConfig) String() string {
 		c.externalConnectivityAddressType,
 		c.redpandaRPCPort,
 		c.hostPort,
-		c.proxyHostPort)
+		c.proxyHostPort,
+		c.dataDirPath)
 }
 
-var errorMissingEnvironmentVariable = errors.New("missing environment variable")
+var errorMissingEnvironmentVariable = fmt.Errorf("missing environment variable")
 
 func main() {
 	log.Print("The redpanda configurator is starting")
@@ -109,32 +129,46 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
-	hostIndex, err := hostIndex(c.hostName)
+	podOrdinal, err := hostIndex(c.hostName)
 	if err != nil {
 		log.Fatalf("%s", fmt.Errorf("unable to extract host index: %w", err))
 	}
 
-	log.Printf("Host index calculated %d", hostIndex)
+	log.Printf("Host index calculated %d", podOrdinal)
 
-	err = registerAdvertisedKafkaAPI(&c, cfg, hostIndex, kafkaAPIPort)
+	err = registerAdvertisedKafkaAPI(&c, cfg, podOrdinal, kafkaAPIPort)
 	if err != nil {
 		log.Fatalf("%s", fmt.Errorf("unable to register advertised Kafka API: %w", err))
 	}
 
 	if cfg.Pandaproxy != nil && len(cfg.Pandaproxy.PandaproxyAPI) > 0 {
 		proxyAPIPort := getInternalProxyAPIPort(cfg)
-		err = registerAdvertisedPandaproxyAPI(&c, cfg, hostIndex, proxyAPIPort)
+		err = registerAdvertisedPandaproxyAPI(&c, cfg, podOrdinal, proxyAPIPort)
 		if err != nil {
 			log.Fatalf("%s", fmt.Errorf("unable to register advertised Pandaproxy API: %w", err))
 		}
 	}
 
-	cfg.Redpanda.ID = int(hostIndex)
+	restCfg, err := k8sConfig.GetConfig()
+	if err != nil {
+		log.Fatalf("%s", fmt.Errorf("unable to create kubernetes rest config: %w", err))
+	}
 
-	// First Redpanda node need to have cleared seed servers in order
-	// to form raft group 0
-	if hostIndex == 0 {
-		cfg.Redpanda.SeedServers = []config.SeedServer{}
+	k8sClient, err := client.New(restCfg, client.Options{})
+	if err != nil {
+		log.Fatalf("%s", fmt.Errorf("unable to create kubernetes client: %w", err))
+	}
+
+	err = calculateRedpandaID(func() int {
+		return int(rand.NewSource(time.Now().Unix()).Int63())
+	}, cfg, c, podOrdinal, k8sClient)
+	if err != nil {
+		log.Fatalf("%s", fmt.Errorf("unable to register node Redpanda ID: %w", err))
+	}
+
+	err = initializeSeedSeverList(cfg, c, podOrdinal, k8sClient)
+	if err != nil {
+		log.Fatalf("%s", fmt.Errorf("unable to determine seed server list: %w", err))
 	}
 
 	cfgBytes, err := yaml.Marshal(cfg)
@@ -149,7 +183,234 @@ func main() {
 	log.Printf("Configuration saved to: %s", c.configDestination)
 }
 
-var errInternalPortMissing = errors.New("port configration is missing internal port")
+func initializeSeedSeverList(
+	cfg *config.Config, c configuratorConfig, index brokerID, k8sClient client.Client,
+) error {
+	if index != 0 {
+		return nil
+	}
+
+	empty, err := IsRedpandaDataFolderEmpty(c.dataDirPath)
+	if err != nil {
+		return fmt.Errorf("checking Redpanda data folder content (%s): %w", c.dataDirPath, err)
+	}
+
+	if !empty {
+		return nil
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	stsName := renderStsName(c.hostName, index)
+
+	sts := &v1.StatefulSet{}
+	err = k8sClient.Get(ctx, client.ObjectKey{
+		Name: stsName,
+	}, sts)
+	if err != nil {
+		return fmt.Errorf("retrieving statful set: %w", err)
+	}
+
+	if *sts.Spec.Replicas == 1 {
+		cfg.Redpanda.SeedServers = []config.SeedServer{}
+		return nil
+	}
+
+	podList := &corev1.PodList{}
+	err = k8sClient.List(ctx, podList, &client.ListOptions{
+		LabelSelector: k8sLabels.SelectorFromSet(map[string]string{
+			labels.InstanceKey: stsName,
+		}),
+	})
+	if err != nil {
+		return fmt.Errorf("listing available PODs: %w", err)
+	}
+
+	anyRedpandaPodIsRunning := false
+	for _, pod := range podList.Items {
+		if pod.Name == c.hostName {
+			continue
+		}
+		if pod.Status.Phase == corev1.PodRunning {
+			anyRedpandaPodIsRunning = true
+			break
+		}
+	}
+
+	if anyRedpandaPodIsRunning {
+		for _, srv := range cfg.Redpanda.SeedServers {
+			cl := &redpandav1alpha1.Cluster{}
+			err = k8sClient.Get(ctx, client.ObjectKey{
+				Name: stsName,
+			}, cl)
+			if err != nil {
+				return fmt.Errorf("retrieving cluster object: %w", err)
+			}
+
+			scheme := "http"
+			if len(cl.Spec.Configuration.AdminAPI) > 0 && cl.Spec.Configuration.AdminAPI[0].TLS.Enabled {
+				scheme = "https"
+			}
+			// TODO: Implement AdminAPI that has RequireClientAuth set to true (mTLS)
+			request, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("%s://%s:%d/v1/brokers", scheme, srv.Host.Address, srv.Host.Port), nil)
+			if err != nil {
+				return fmt.Errorf("creating seed server broker reques: %w", err)
+			}
+
+			tr := http.DefaultTransport.(*http.Transport).Clone()
+			// TODO: construct client with root CA for AdminApi that should be mounted
+			// to the configurator container.
+			tr.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
+			httpClient := &http.Client{Transport: tr}
+			resp, err := httpClient.Do(request)
+			if err != nil {
+				// init container need to be robust in retrieving other brokers cluster view
+				// any error might block Redpanda initialization
+				continue
+			}
+			b, err := io.ReadAll(resp.Body)
+			resp.Body.Close()
+			if err != nil {
+				return fmt.Errorf("reading seed server broker body: %w", err)
+			}
+
+			var brokers []broker
+			err = json.Unmarshal(b, &brokers)
+			if err != nil {
+				return fmt.Errorf("unmarshalling brokers response: %w", err)
+			}
+
+			for _, br := range brokers {
+				if br.NodeId == 0 {
+					continue
+				}
+				if br.IsAlive {
+					return nil
+				}
+			}
+		}
+	}
+
+	cfg.Redpanda.SeedServers = []config.SeedServer{}
+
+	return nil
+}
+
+type broker struct {
+	NodeId  int  `json:"node_id"`
+	IsAlive bool `json:"is_alive"`
+}
+
+func renderStsName(podName string, index brokerID) string {
+	indexStr := strconv.Itoa(int(index))
+	return podName[:len(podName)-1-len(indexStr)]
+}
+
+func calculateRedpandaID(
+	randID func() int, cfg *config.Config, c configuratorConfig, initialID brokerID, k8sClient client.Client,
+) error {
+	redpandaIDFile := filepath.Join(c.dataDirPath, ".redpanda_id")
+
+	_, err := os.Stat(redpandaIDFile)
+	if errors.Is(err, os.ErrNotExist) {
+		empty, err := IsRedpandaDataFolderEmpty(c.dataDirPath)
+		if err != nil {
+			return fmt.Errorf("checking Redpanda data folder content (%s): %w", c.dataDirPath, err)
+		}
+
+		if !empty {
+			err = os.WriteFile(redpandaIDFile, []byte(strconv.Itoa(int(initialID))), 0o666)
+			if err != nil {
+				return fmt.Errorf("storing redpanda ID in data folder (%s): %w", c.dataDirPath, err)
+			}
+			cfg.Redpanda.ID = int(initialID)
+			return nil
+		}
+
+		redpandaID := randID()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+
+		if len(c.hostName) < 3 {
+			return fmt.Errorf("hostname should have more than 3 caracters")
+		}
+		cmName := renderStsName(c.hostName, initialID)
+
+		cm := &corev1.ConfigMap{}
+		err = k8sClient.Get(ctx, client.ObjectKey{
+			Name: fmt.Sprintf("%s-ids", cmName),
+		}, cm)
+		if err != nil && k8sErrors.IsNotFound(err) {
+			cm.Name = fmt.Sprintf("%s-ids", cmName)
+			cm.Data = map[string]string{
+				redpandaIDs: "",
+			}
+			err = k8sClient.Create(ctx, cm)
+			if err != nil {
+				return fmt.Errorf("creating Redpanda ID in configmap: %w", err)
+			}
+		} else if err != nil {
+			return fmt.Errorf("retreving redpanda ids config map: %w", err)
+		}
+
+		ids, ok := cm.Data[redpandaIDs]
+		if ok {
+			splitIDs := strings.Split(ids, ";")
+			found := true
+			for found {
+				found = false
+				for _, id := range splitIDs {
+					if id == strconv.Itoa(redpandaID) {
+						redpandaID = randID()
+						found = true
+						break
+					}
+				}
+			}
+
+			cm.Data[redpandaIDs] = fmt.Sprintf("%s;%d", ids, redpandaID)
+			err = k8sClient.Update(ctx, cm)
+			if err != nil {
+				return fmt.Errorf("updating redpanda ids config map: %w", err)
+			}
+		}
+
+		err = os.WriteFile(redpandaIDFile, []byte(strconv.Itoa(redpandaID)), 0o666)
+		if err != nil {
+			return fmt.Errorf("storing redpanda ID in data folder (%s): %w", c.dataDirPath, err)
+		}
+		cfg.Redpanda.ID = redpandaID
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("stat redpanda ID file: %w", err)
+	}
+
+	redpandaID, err := os.ReadFile(redpandaIDFile)
+	if err != nil {
+		return fmt.Errorf("reading redpanda id file: %w", err)
+	}
+
+	rID, err := strconv.Atoi(string(redpandaID))
+	if err != nil {
+		return fmt.Errorf("converting content of redpanda id file to int: %w", err)
+	}
+	cfg.Redpanda.ID = rID
+
+	return nil
+}
+
+func IsRedpandaDataFolderEmpty(dataDirPath string) (bool, error) {
+	de, err := os.ReadDir(dataDirPath)
+	if err != nil {
+		return false, fmt.Errorf("reading volume content: %w", err)
+	}
+	return len(de) == 0, nil
+}
+
+var errInternalPortMissing = fmt.Errorf("port configration is missing internal port")
 
 func getInternalKafkaAPIPort(cfg *config.Config) (int, error) {
 	for _, l := range cfg.Redpanda.KafkaAPI {
@@ -336,6 +597,10 @@ func checkEnvVars() (configuratorConfig, error) {
 			value: &hostPort,
 			name:  hostPortEnvVar,
 		},
+		{
+			value: &c.dataDirPath,
+			name:  dataDirPath,
+		},
 	}
 	for _, envVar := range envVarList {
 		v, exist := os.LookupEnv(envVar.name)
@@ -385,8 +650,7 @@ func checkEnvVars() (configuratorConfig, error) {
 }
 
 // hostIndex takes advantage of pod naming convention in Kubernetes StatfulSet
-// the last number is the index of replica. This index is then propagated
-// to redpanda.node_id.
+// the last number is the index of replica.
 func hostIndex(hostName string) (brokerID, error) {
 	s := strings.Split(hostName, "-")
 	last := len(s) - 1