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,66 @@
+package cassdcutil
+
+import (
+ "context"
+ "fmt"
+
+ cassdcapi "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
+ appsv1 "k8s.io/api/apps/v1"
+ 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
+// We use the OwnerReference method because the pod labels are incorrect if datacenter name override is used
+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 len(pod.OwnerReferences) < 1 {
+ return nil, fmt.Errorf("target pod not managed by cass-operator, no owner reference")
+ }
+
+ statefulSet := &appsv1.StatefulSet{}
+ err = c.client.Get(ctx, types.NamespacedName{Namespace: namespace, Name: pod.OwnerReferences[0].Name}, statefulSet)
+ if err != nil {
+ return nil, err
+ }
+
+ if len(statefulSet.OwnerReferences) < 1 {
+ return nil, fmt.Errorf("target statefulset not managed by cass-operator, no owner reference")
+ }
+
+ cassDcKey := types.NamespacedName{Namespace: namespace, Name: statefulSet.OwnerReferences[0].Name}
+ 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
+}