configmap.go

package cassandradatacenter

import (
	"context"
	"fmt"
	cassandraoperatorv1alpha1 "github.com/instaclustr/cassandra-operator/pkg/apis/cassandraoperator/v1alpha1"
	"gopkg.in/yaml.v2"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"regexp"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"strings"
)

const (
	MEBIBYTE = 1 << 20
	GIBIBYTE = 1 << 30
)

func createOrUpdateOperatorConfigMap(rctx *reconciliationRequestContext, seedNodesService *corev1.Service) (*corev1.Volume, error) {
	configMap := &corev1.ConfigMap{ObjectMeta: DataCenterResourceMetadata(rctx.cdc, "operator-config")}

	logger := rctx.logger.WithValues("ConfigMap.Name", configMap.Name)

	volumeSource := &corev1.ConfigMapVolumeSource{LocalObjectReference: corev1.LocalObjectReference{Name: configMap.Name}}

	opresult, err := controllerutil.CreateOrUpdate(context.TODO(), rctx.client, configMap, func(_ runtime.Object) error {
		addFileFn := func(path string, data string) {
			configMapVolumeAddTextFile(configMap, volumeSource, path, data)
		}

		addCassandraYamlOverrides(rctx.cdc, seedNodesService, addFileFn)

		addCassandraGossipingPropertyFileSnitchProperties(rctx.cdc, addFileFn)

		addCassandraJVMOptions(rctx.cdc, addFileFn)

		addPrometheusSupport(rctx.cdc, addFileFn)


		if err := controllerutil.SetControllerReference(rctx.cdc, configMap, rctx.scheme); err != nil {
			return err
		}

		return nil
	})

	if err != nil {
		return nil, err
	}

	logger.Info(fmt.Sprintf("ConfigMap %s.", opresult))

	configVolume := &corev1.Volume{
		Name:         "operator-config-volume",
		VolumeSource: corev1.VolumeSource{ConfigMap: volumeSource},
	}

	return configVolume, nil
}

func addCassandraGossipingPropertyFileSnitchProperties(cdc *cassandraoperatorv1alpha1.CassandraDataCenter, addFileFn func(path string, data string)) {
	var writer strings.Builder

	_, _ = fmt.Fprintln(&writer, "# generated by cassandra-operator")

	properties := map[string]string{
		"dc":           cdc.Name,
		"rack":         "rack1",
		"prefer_local": "true",
	}

	writeProperty := func(key string, value string) {
		_, _ = fmt.Fprintf(&writer, "%s=%s\n", key, value)
	}

	for key, value := range properties {
		writeProperty(key, value)
	}

	addFileFn("cassandra-rackdc.properties", writer.String())
}

func addPrometheusSupport(cdc *cassandraoperatorv1alpha1.CassandraDataCenter, addFileFn func(path string, data string)) {
	if cdc.Spec.PrometheusSupport {
		addFileFn(
			"cassandra-env.sh.d/001-cassandra-exporter.sh",
			"JVM_OPTS=\"${JVM_OPTS} -javaagent:${CASSANDRA_HOME}/agents/cassandra-exporter-agent.jar=@${CASSANDRA_CONF}/cassandra-exporter.conf\"",
		)
	}
}

func addCassandraYamlOverrides(cdc *cassandraoperatorv1alpha1.CassandraDataCenter, seedNodesService *corev1.Service, addFileFn func(path string, data string)) {
	type SeedProvider struct {
		ClassName  string              `yaml:"class_name"`
		Parameters []map[string]string `yaml:"parameters"`
	}

	type CassandraConfig struct {
		ClusterName   string  `yaml:"cluster_name"`
		ListenAddress *string `yaml:"listen_address"`
		RPCAddress    *string `yaml:"rpc_address"`

		SeedProvider []SeedProvider `yaml:"seed_provider"`

		EndpointSnitch string `yaml:"endpoint_snitch"`
	}

	data, err := yaml.Marshal(&CassandraConfig{
		ClusterName: cdc.Spec.Cluster,

		ListenAddress: nil, // let C* discover the listen address
		RPCAddress:    nil, // let C* discover the rpc address

		SeedProvider: []SeedProvider{
			{
				ClassName: "com.instaclustr.cassandra.k8s.SeedProvider",
				Parameters: []map[string]string{
					{"service": seedNodesService.Name},
				},
			},
		},

		EndpointSnitch: "org.apache.cassandra.locator.GossipingPropertyFileSnitch", // TODO: custom snitch implementation?
	})

	if err != nil {
		// we're serializing a known structure to YAML -- if that fails...
		panic(err)
	}

	addFileFn("cassandra.yaml.d/001-operator-overrides.yaml", string(data))
}

func addCassandraJVMOptions(cdc *cassandraoperatorv1alpha1.CassandraDataCenter, addFileFn func(path string, data string)) {
	// TODO: should this be Limits or Requests?
	memoryLimit := cdc.Spec.Resources.Limits.Memory().Value()

	jvmHeapSize := maxInt64(minInt64(memoryLimit/2, GIBIBYTE), minInt64(memoryLimit/4, 8*GIBIBYTE))

	youngGenSize := youngGen(jvmHeapSize)

	useG1GC := jvmHeapSize > 8*GIBIBYTE

	var writer strings.Builder

	_, _ = fmt.Fprintf(&writer, "-Xms%d\n", jvmHeapSize) // min heap size
	_, _ = fmt.Fprintf(&writer, "-Xmx%d\n", jvmHeapSize) // max heap size

	// copied from stock jvm.options
	if !useG1GC {
		_, _ = fmt.Fprintf(&writer, "-Xmn%d\n", youngGenSize) // young gen size

		_, _ = fmt.Fprintln(&writer, "-XX:+UseParNewGC")
		_, _ = fmt.Fprintln(&writer, "-XX:+UseConcMarkSweepGC")
		_, _ = fmt.Fprintln(&writer, "-XX:+CMSParallelRemarkEnabled")
		_, _ = fmt.Fprintln(&writer, "-XX:SurvivorRatio=8")
		_, _ = fmt.Fprintln(&writer, "-XX:MaxTenuringThreshold=1")
		_, _ = fmt.Fprintln(&writer, "-XX:CMSInitiatingOccupancyFraction=75")
		_, _ = fmt.Fprintln(&writer, "-XX:+UseCMSInitiatingOccupancyOnly")
		_, _ = fmt.Fprintln(&writer, "-XX:CMSWaitDuration=10000")
		_, _ = fmt.Fprintln(&writer, "-XX:+CMSParallelInitialMarkEnabled")
		_, _ = fmt.Fprintln(&writer, "-XX:+CMSEdenChunksRecordAlways")
		_, _ = fmt.Fprintln(&writer, "-XX:+CMSClassUnloadingEnabled")

	} else {
		_, _ = fmt.Fprintln(&writer, "-XX:+UseG1GC")
		_, _ = fmt.Fprintln(&writer, "-XX:G1RSetUpdatingPauseTimePercent=5")
		_, _ = fmt.Fprintln(&writer, "-XX:MaxGCPauseMillis=500")

		if jvmHeapSize > 16*GIBIBYTE {
			_, _ = fmt.Fprintln(&writer, "-XX:InitiatingHeapOccupancyPercent=70")
		}

		// TODO: tune -XX:ParallelGCThreads, -XX:ConcGCThreads
	}

	// OOM Error handling
	_, _ = fmt.Fprintln(&writer, "-XX:+HeapDumpOnOutOfMemoryError")
	_, _ = fmt.Fprintln(&writer, "-XX:+CrashOnOutOfMemoryError")

	// TODO: maybe tune -Dcassandra.available_processors=number_of_processors - Wait till we build C* for Java 11
	// not sure if k8s exposes the right number of CPU cores inside the container

	addFileFn("jvm.options.d/001-jvm-memory-gc.options", writer.String())
}

func youngGen(jvmHeapSize int64) int64 {

	coreCount := int64(4) // TODO

	return minInt64(coreCount*MEBIBYTE, jvmHeapSize/4)
}

func minInt64(a int64, b int64) int64 {
	if a < b {
		return a
	}
	return b
}

func maxInt64(a int64, b int64) int64 {
	if a > b {
		return a
	}
	return b
}

func configMapVolumeAddTextFile(configMap *corev1.ConfigMap, volumeSource *corev1.ConfigMapVolumeSource, path string, data string) {
	encodedKey := regexp.MustCompile("\\W").ReplaceAllLiteralString(path, "_")

	// lazy init
	if configMap.Data == nil {
		configMap.Data = make(map[string]string)
	}

	configMap.Data[encodedKey] = data
	volumeSource.Items = append(volumeSource.Items, corev1.KeyToPath{Key: encodedKey, Path: path})
}