Add nodetool command for pod access with support for automated TLS

Makefile

@@ -19,7 +19,7 @@ ENVTEST_K8S_VERSION = 1.28.x
 GO_FLAGS ?= -v
 
 .PHONY: all
-all: build
+all: test build
 
 ##@ General
 
@@ -59,7 +59,7 @@ lint: golangci-lint ## Run golangci-lint against code
 	$(GOLANGCI_LINT) run ./...
 
 .PHONY: build
-build: test ## Build kubectl-k8ssandra
+build: ## Build kubectl-k8ssandra
 	CGO_ENABLED=0 go build -o kubectl-k8ssandra cmd/kubectl-k8ssandra/main.go
 
 .PHONY: docker-build

cmd/kubectl-k8ssandra/k8ssandra/k8ssandra.go

@@ -7,9 +7,9 @@ import (
 	// "github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/edit"
 	// "github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/list"
 	// "github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/migrate"
-	// "github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/nodetool"
 	"github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/config"
 	"github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/helm"
+	"github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/nodetool"
 	"github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/operate"
 	"github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/register"
 	"github.com/k8ssandra/k8ssandra-client/cmd/kubectl-k8ssandra/users"
@@ -54,6 +54,7 @@ func NewCmd(streams genericclioptions.IOStreams) *cobra.Command {
 	// cmd.AddCommand(migrate.NewInstallCmd(streams))
 	cmd.AddCommand(config.NewCmd(streams))
 	cmd.AddCommand(helm.NewHelmCmd(streams))
+	cmd.AddCommand(nodetool.NewCmd(streams))
 	register.SetupRegisterClusterCmd(cmd, streams)
 
 	// cmd.Flags().BoolVar(&o.listNamespaces, "list", o.listNamespaces, "if true, print the list of all namespaces in the current KUBECONFIG")

cmd/kubectl-k8ssandra/nodetool/nodetool.go

@@ -0,0 +1,140 @@
+package nodetool
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/k8ssandra/k8ssandra-client/pkg/cassdcutil"
+	"github.com/k8ssandra/k8ssandra-client/pkg/kubernetes"
+	"github.com/k8ssandra/k8ssandra-client/pkg/util"
+	"github.com/spf13/cobra"
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+	"k8s.io/kubectl/pkg/cmd/exec"
+)
+
+var (
+	cqlshExample = `
+	# launch a interactive cqlsh shell on node
+	%[1]s nodetool <pod> <command> [<args>]
+`
+	errNotEnoughParameters = fmt.Errorf("not enough parameters to run nodetool")
+)
+
+type options struct {
+	configFlags *genericclioptions.ConfigFlags
+	genericclioptions.IOStreams
+	execOptions *exec.ExecOptions
+	cassManager *cassdcutil.CassManager
+	params      []string
+}
+
+func newOptions(streams genericclioptions.IOStreams) *options {
+	return &options{
+		configFlags: genericclioptions.NewConfigFlags(true),
+		IOStreams:   streams,
+	}
+}
+
+// NewCmd provides a cobra command wrapping cqlShOptions
+func NewCmd(streams genericclioptions.IOStreams) *cobra.Command {
+	o := newOptions(streams)
+
+	cmd := &cobra.Command{
+		Use:          "nodetool [pod] [flags]",
+		Short:        "nodetool launched on pod",
+		Example:      fmt.Sprintf(cqlshExample, "kubectl k8ssandra"),
+		SilenceUsage: true,
+		RunE: func(c *cobra.Command, args []string) error {
+			if err := o.Complete(c, args); err != nil {
+				return err
+			}
+			if err := o.Validate(); err != nil {
+				return err
+			}
+			if err := o.Run(); err != nil {
+				return err
+			}
+
+			return nil
+		},
+	}
+
+	o.configFlags.AddFlags(cmd.Flags())
+	return cmd
+}
+
+// Complete parses the arguments and necessary flags to options
+func (c *options) Complete(cmd *cobra.Command, args []string) error {
+	var err error
+
+	if len(args) < 2 {
+		return errNotEnoughParameters
+	}
+
+	execOptions, err := util.GetExecOptions(c.IOStreams, c.configFlags)
+	if err != nil {
+		return err
+	}
+	c.execOptions = execOptions
+	execOptions.PodName = args[0]
+
+	restConfig, err := c.configFlags.ToRESTConfig()
+	if err != nil {
+		return err
+	}
+
+	kubeClient, err := kubernetes.GetClientInNamespace(restConfig, execOptions.Namespace)
+	if err != nil {
+		return err
+	}
+
+	c.cassManager = cassdcutil.NewManager(kubeClient)
+
+	c.params = args[1:]
+
+	return nil
+}
+
+// Validate ensures that all required arguments and flag values are provided
+func (c *options) Validate() error {
+	// We could validate here if a nodetool command requires flags, but lets let nodetool throw that error
+
+	return nil
+}
+
+// Run triggers the nodetool command on target pod
+func (c *options) Run() error {
+	ctx := context.Background()
+
+	dc, err := c.cassManager.PodDatacenter(ctx, c.execOptions.PodName, c.execOptions.Namespace)
+	if err != nil {
+		return err
+	}
+
+	cassSecret, err := c.cassManager.CassandraAuthDetails(ctx, dc)
+	if err != nil {
+		return err
+	}
+	c.execOptions.Command = []string{"nodetool"}
+
+	c.execOptions.Command = append(c.execOptions.Command, nodetoolAuthParameters(cassSecret)...)
+
+	c.execOptions.Command = append(c.execOptions.Command, c.params...)
+
+	return c.execOptions.Run()
+}
+
+func nodetoolAuthParameters(authDetails *cassdcutil.CassandraAuth) []string {
+	auth := []string{"--username", authDetails.Username, "--password", authDetails.Password}
+
+	if authDetails.KeystorePath != "" {
+		auth = append(auth, "-Dcom.sun.management.jmxremote.ssl.need.client.auth=true")
+		auth = append(auth, "-Dcom.sun.management.jmxremote.registry.ssl=true")
+		auth = append(auth, "-Djavax.net.ssl.keyStore="+authDetails.KeystorePath)
+		auth = append(auth, "-Djavax.net.ssl.keyStorePassword="+authDetails.KeystorePassword)
+		auth = append(auth, "-Djavax.net.ssl.trustStore="+authDetails.TruststorePath)
+		auth = append(auth, "-Djavax.net.ssl.trustStorePassword="+authDetails.TruststorePassword)
+	}
+
+	return auth
+}

pkg/cassdcutil/config.go

@@ -0,0 +1,34 @@
+package cassdcutil
+
+import (
+	"github.com/Jeffail/gabs/v2"
+	cassdcapi "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
+)
+
+func ClientEncryptionEnabled(dc *cassdcapi.CassandraDatacenter) bool {
+	config, err := gabs.ParseJSON(dc.Spec.Config)
+	if err != nil {
+		return false
+	}
+
+	if config.Exists("cassandra-yaml", "client_encryption_options") {
+		if config.Path("cassandra-yaml.client_encryption_options.enabled").Data().(bool) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func SubSectionOfCassYaml(dc *cassdcapi.CassandraDatacenter, section string) map[string]*gabs.Container {
+	config, err := gabs.ParseJSON(dc.Spec.Config)
+	if err != nil {
+		return make(map[string]*gabs.Container)
+	}
+
+	if !config.Exists("cassandra-yaml") {
+		return make(map[string]*gabs.Container)
+	}
+
+	return config.Path("cassandra-yaml").Path(section).ChildrenMap()
+}

pkg/cassdcutil/config_test.go

@@ -0,0 +1,101 @@
+package cassdcutil
+
+import (
+	"encoding/json"
+	"testing"
+
+	cassdcapi "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
+	"github.com/stretchr/testify/assert"
+)
+
+var clientEncryptionEnabled = `
+{
+	"cassandra-yaml": {
+	  "client_encryption_options": {
+	  	"enabled": true,
+		"optional": false,
+        "keystore": "/etc/encryption/node-keystore.jks",
+        "keystore_password": "dc2",
+        "truststore": "/etc/encryption/node-keystore.jks",
+        "truststore_password": "dc2"
+	  }
+	}
+}
+`
+
+func TestClientEncryptionEnabled(t *testing.T) {
+	dc := &cassdcapi.CassandraDatacenter{
+		Spec: cassdcapi.CassandraDatacenterSpec{
+			Config: json.RawMessage(clientEncryptionEnabled),
+		},
+	}
+
+	assert := assert.New(t)
+	assert.True(ClientEncryptionEnabled(dc))
+}
+
+func TestEmptySubSection(t *testing.T) {
+	dc := &cassdcapi.CassandraDatacenter{
+		Spec: cassdcapi.CassandraDatacenterSpec{},
+	}
+
+	assert := assert.New(t)
+	section := SubSectionOfCassYaml(dc, "client_encryption_options")
+	assert.NotNil(section)
+	assert.Equal(0, len(section))
+
+	dc.Spec.Config = json.RawMessage(``)
+	section = SubSectionOfCassYaml(dc, "client_encryption_options")
+	assert.NotNil(section)
+	assert.Equal(0, len(section))
+}
+
+func TestSubSectionNotMatch(t *testing.T) {
+	dc := &cassdcapi.CassandraDatacenter{
+		Spec: cassdcapi.CassandraDatacenterSpec{
+			Config: json.RawMessage(clientEncryptionEnabled),
+		},
+	}
+
+	assert := assert.New(t)
+	section := SubSectionOfCassYaml(dc, "server_encryption_options")
+	assert.NotNil(section)
+	assert.Equal(0, len(section))
+}
+
+func TestSubSectionPart(t *testing.T) {
+	dc := &cassdcapi.CassandraDatacenter{
+		Spec: cassdcapi.CassandraDatacenterSpec{
+			Config: json.RawMessage(clientEncryptionEnabled),
+		},
+	}
+
+	assert := assert.New(t)
+	section := SubSectionOfCassYaml(dc, "client_encryption_options")
+	assert.NotNil(section)
+	assert.Equal(6, len(section))
+
+	enabled, ok := section["enabled"].Data().(bool)
+	assert.True(ok)
+	assert.True(enabled)
+
+	keystore, ok := section["keystore"].Data().(string)
+	assert.True(ok)
+	assert.Equal("/etc/encryption/node-keystore.jks", keystore)
+
+	keystorePassword, ok := section["keystore_password"].Data().(string)
+	assert.True(ok)
+	assert.Equal("dc2", keystorePassword)
+
+	truststore, ok := section["truststore"].Data().(string)
+	assert.True(ok)
+	assert.Equal("/etc/encryption/node-keystore.jks", truststore)
+
+	truststorePassword, ok := section["truststore_password"].Data().(string)
+	assert.True(ok)
+	assert.Equal("dc2", truststorePassword)
+
+	optional, ok := section["optional"].Data().(bool)
+	assert.True(ok)
+	assert.False(optional)
+}

pkg/cassdcutil/fetcher.go

@@ -0,0 +1,55 @@
+package cassdcutil
+
+import (
+	"context"
+	"fmt"
+
+	cassdcapi "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// CassandraDatacenter fetches the CassandraDatacenter by its name and namespace
+func (c *CassManager) CassandraDatacenter(ctx context.Context, name, namespace string) (*cassdcapi.CassandraDatacenter, error) {
+	cassdcKey := types.NamespacedName{Namespace: namespace, Name: name}
+	cassdc := &cassdcapi.CassandraDatacenter{}
+
+	if err := c.client.Get(ctx, cassdcKey, cassdc); err != nil {
+		return nil, err
+	}
+
+	return cassdc, nil
+}
+
+// PodDatacenter returns the CassandraDatacenter instance of the pod if it's managed by cass-operator
+func (c *CassManager) PodDatacenter(ctx context.Context, podName, namespace string) (*cassdcapi.CassandraDatacenter, error) {
+	key := types.NamespacedName{Namespace: namespace, Name: podName}
+	pod := &corev1.Pod{}
+	err := c.client.Get(ctx, key, pod)
+	if err != nil {
+		return nil, err
+	}
+
+	if dc, found := pod.Labels[cassdcapi.DatacenterLabel]; !found {
+		return nil, fmt.Errorf("target pod not managed by cass-operator, no datacenter label")
+	} else {
+		// Get CassandraDatacenter for the dc
+		cassDcKey := types.NamespacedName{Namespace: namespace, Name: dc}
+		cassdc := &cassdcapi.CassandraDatacenter{}
+		err = c.client.Get(ctx, cassDcKey, cassdc)
+		if err != nil {
+			return nil, err
+		}
+
+		return cassdc, nil
+	}
+}
+
+// CassandraDatacenterPods returns the pods of the CassandraDatacenter
+func (c *CassManager) CassandraDatacenterPods(ctx context.Context, cassdc *cassdcapi.CassandraDatacenter) (*corev1.PodList, error) {
+	// What if same namespace has two datacenters with the same name? Can that happen?
+	podList := &corev1.PodList{}
+	err := c.client.List(ctx, podList, client.InNamespace(cassdc.Namespace), client.MatchingLabels(map[string]string{cassdcapi.DatacenterLabel: cassdc.Name}))
+	return podList, err
+}