Travis Nielsen
d0c7d28a63
The operator kit had more utility originally when the operator was creating and managing the TPRs and CRDs directly. Since the CRDs are now created from a manifest and no longer by the operators, the utility of operator kit is limited to the controller watcher. Since we are moving to the controller runtime we simplify the code to make the transition smoother. Now there is only a simple WatchCR method that will need to be replaced as we maek that transition. Signed-off-by: Travis Nielsen <tnielsen@redhat.com>
414 lines
12 KiB
Go
414 lines
12 KiB
Go
/*
|
|
Copyright 2019 The Rook Authors. All rights reserved.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
package yugabytedb
|
|
|
|
import (
|
|
"fmt"
|
|
"reflect"
|
|
|
|
rookv1alpha2 "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
|
|
yugabytedbv1alpha1 "github.com/rook/rook/pkg/apis/yugabytedb.rook.io/v1alpha1"
|
|
"github.com/rook/rook/pkg/clusterd"
|
|
"github.com/rook/rook/pkg/operator/k8sutil"
|
|
v1 "k8s.io/api/core/v1"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"k8s.io/apimachinery/pkg/util/intstr"
|
|
"k8s.io/client-go/tools/cache"
|
|
)
|
|
|
|
const (
|
|
customResourceName = "ybcluster"
|
|
customResourceNamePlural = "ybclusters"
|
|
masterName = "yb-master"
|
|
masterNamePlural = "yb-masters"
|
|
tserverName = "yb-tserver"
|
|
tserverNamePlural = "yb-tservers"
|
|
masterUIServiceName = "yb-master-ui"
|
|
tserverUIServiceName = "yb-tserver-ui"
|
|
masterUIPortDefault = int32(7000)
|
|
masterUIPortName = "yb-master-ui"
|
|
masterRPCPortDefault = int32(7100)
|
|
masterRPCPortName = "yb-master-rpc"
|
|
tserverUIPortDefault = int32(9000)
|
|
tserverUIPortName = "yb-tserver-ui"
|
|
tserverRPCPortDefault = int32(9100)
|
|
tserverRPCPortName = "yb-tserver-rpc"
|
|
tserverCassandraPortDefault = int32(9042)
|
|
tserverCassandraPortName = "ycql"
|
|
tserverRedisPortDefault = int32(6379)
|
|
tserverRedisPortName = "yedis"
|
|
tserverPostgresPortDefault = int32(5433)
|
|
tserverPostgresPortName = "ysql"
|
|
masterContainerUIPortName = "master-ui"
|
|
masterContainerRPCPortName = "master-rpc"
|
|
tserverContainerUIPortName = "tserver-ui"
|
|
tserverContainerRPCPortName = "tserver-rpc"
|
|
uiPortName = "ui"
|
|
rpcPortName = "rpc-port"
|
|
cassandraPortName = "cassandra"
|
|
redisPortName = "redis"
|
|
postgresPortName = "postgres"
|
|
volumeMountPath = "/mnt/data0"
|
|
envGetHostsFrom = "GET_HOSTS_FROM"
|
|
envGetHostsFromVal = "dns"
|
|
envPodIP = "POD_IP"
|
|
envPodIPVal = "status.podIP"
|
|
envPodName = "POD_NAME"
|
|
envPodNameVal = "metadata.name"
|
|
yugabyteDBImageName = "yugabytedb/yugabyte:1.3.1.0-b16"
|
|
)
|
|
|
|
var ClusterResource = k8sutil.CustomResource{
|
|
Name: customResourceName,
|
|
Plural: customResourceNamePlural,
|
|
Group: yugabytedbv1alpha1.CustomResourceGroup,
|
|
Version: yugabytedbv1alpha1.Version,
|
|
Kind: reflect.TypeOf(yugabytedbv1alpha1.YBCluster{}).Name(),
|
|
}
|
|
|
|
type ClusterController struct {
|
|
context *clusterd.Context
|
|
containerImage string
|
|
}
|
|
|
|
func NewClusterController(context *clusterd.Context, containerImage string) *ClusterController {
|
|
return &ClusterController{
|
|
context: context,
|
|
containerImage: containerImage,
|
|
}
|
|
}
|
|
|
|
type cluster struct {
|
|
context *clusterd.Context
|
|
name string
|
|
namespace string
|
|
spec yugabytedbv1alpha1.YBClusterSpec
|
|
annotations rookv1alpha2.Annotations
|
|
ownerRef metav1.OwnerReference
|
|
}
|
|
|
|
type clusterPorts struct {
|
|
masterPorts, tserverPorts serverPorts
|
|
}
|
|
|
|
type serverPorts struct {
|
|
ui, rpc, cassandra, redis, postgres int32
|
|
}
|
|
|
|
func NewCluster(c *yugabytedbv1alpha1.YBCluster, context *clusterd.Context) *cluster {
|
|
return &cluster{
|
|
context: context,
|
|
name: c.Name,
|
|
namespace: c.Namespace,
|
|
spec: c.Spec,
|
|
annotations: c.Spec.Annotations,
|
|
ownerRef: clusterOwnerRef(c.Name, string(c.UID)),
|
|
}
|
|
}
|
|
|
|
func clusterOwnerRef(name, clusterID string) metav1.OwnerReference {
|
|
blockOwner := true
|
|
return metav1.OwnerReference{
|
|
APIVersion: fmt.Sprintf("%s/%s", ClusterResource.Group, ClusterResource.Version),
|
|
Kind: ClusterResource.Kind,
|
|
Name: name,
|
|
UID: types.UID(clusterID),
|
|
BlockOwnerDeletion: &blockOwner,
|
|
}
|
|
}
|
|
|
|
func (c *ClusterController) StartWatch(namespace string, stopCh chan struct{}) error {
|
|
resourceHandlerFuncs := cache.ResourceEventHandlerFuncs{
|
|
AddFunc: c.OnAdd,
|
|
UpdateFunc: c.OnUpdate,
|
|
DeleteFunc: c.onDelete,
|
|
}
|
|
|
|
logger.Infof("start watching yugabytedb clusters in all namespaces")
|
|
go k8sutil.WatchCR(ClusterResource, namespace, resourceHandlerFuncs, c.context.RookClientset.YugabytedbV1alpha1().RESTClient(), &yugabytedbv1alpha1.YBCluster{}, stopCh)
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *ClusterController) onDelete(obj interface{}) {
|
|
cluster, ok := obj.(*yugabytedbv1alpha1.YBCluster)
|
|
if !ok {
|
|
return
|
|
}
|
|
cluster = cluster.DeepCopy()
|
|
logger.Infof("cluster %s deleted from namespace %s", cluster.Name, cluster.Namespace)
|
|
}
|
|
|
|
func validateClusterSpec(spec yugabytedbv1alpha1.YBClusterSpec) error {
|
|
|
|
if spec.Master.Replicas < 1 {
|
|
return fmt.Errorf("invalid Master replica count: %d. Must be at least 1", spec.Master.Replicas)
|
|
}
|
|
|
|
if spec.TServer.Replicas < 1 {
|
|
return fmt.Errorf("invalid TServer replica count: %d. Must be at least 1", spec.TServer.Replicas)
|
|
}
|
|
|
|
if _, err := getPortsFromSpec(spec.Master.Network); err != nil {
|
|
return err
|
|
}
|
|
|
|
if _, err := getPortsFromSpec(spec.TServer.Network); err != nil {
|
|
return err
|
|
}
|
|
|
|
if &spec.Master.VolumeClaimTemplate == nil {
|
|
return fmt.Errorf("VolumeClaimTemplate unavailable in Master spec.")
|
|
}
|
|
|
|
if &spec.TServer.VolumeClaimTemplate == nil {
|
|
return fmt.Errorf("VolumeClaimTemplate unavailable in TServer spec.")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func createAppLabels(label string) map[string]string {
|
|
return map[string]string{
|
|
k8sutil.AppAttr: label,
|
|
}
|
|
}
|
|
|
|
func createServicePorts(cluster *cluster, isTServerService bool) []v1.ServicePort {
|
|
var servicePorts []v1.ServicePort
|
|
|
|
if !isTServerService {
|
|
ports, _ := getPortsFromSpec(cluster.spec.Master.Network)
|
|
|
|
servicePorts = []v1.ServicePort{
|
|
{
|
|
Name: uiPortName,
|
|
Port: ports.masterPorts.ui,
|
|
TargetPort: intstr.FromInt(int(ports.masterPorts.ui)),
|
|
},
|
|
{
|
|
Name: rpcPortName,
|
|
Port: ports.masterPorts.rpc,
|
|
TargetPort: intstr.FromInt(int(ports.masterPorts.rpc)),
|
|
},
|
|
}
|
|
} else {
|
|
ports, _ := getPortsFromSpec(cluster.spec.TServer.Network)
|
|
|
|
tserverUIPort := ports.tserverPorts.ui
|
|
|
|
if tserverUIPort <= 0 {
|
|
tserverUIPort = tserverUIPortDefault
|
|
}
|
|
|
|
servicePorts = []v1.ServicePort{
|
|
{
|
|
Name: uiPortName,
|
|
Port: tserverUIPort,
|
|
TargetPort: intstr.FromInt(int(tserverUIPort)),
|
|
},
|
|
{
|
|
Name: rpcPortName,
|
|
Port: ports.tserverPorts.rpc,
|
|
TargetPort: intstr.FromInt(int(ports.tserverPorts.rpc)),
|
|
},
|
|
{
|
|
Name: cassandraPortName,
|
|
Port: ports.tserverPorts.cassandra,
|
|
TargetPort: intstr.FromInt(int(ports.tserverPorts.cassandra)),
|
|
},
|
|
{
|
|
Name: redisPortName,
|
|
Port: ports.tserverPorts.redis,
|
|
TargetPort: intstr.FromInt(int(ports.tserverPorts.redis)),
|
|
},
|
|
{
|
|
Name: postgresPortName,
|
|
Port: ports.tserverPorts.postgres,
|
|
TargetPort: intstr.FromInt(int(ports.tserverPorts.postgres)),
|
|
},
|
|
}
|
|
}
|
|
|
|
return servicePorts
|
|
}
|
|
|
|
func createUIServicePorts(ports *clusterPorts, isTServerService bool) []v1.ServicePort {
|
|
var servicePorts []v1.ServicePort
|
|
|
|
if !isTServerService {
|
|
servicePorts = []v1.ServicePort{
|
|
{
|
|
Name: uiPortName,
|
|
Port: ports.masterPorts.ui,
|
|
TargetPort: intstr.FromInt(int(ports.masterPorts.ui)),
|
|
},
|
|
}
|
|
} else {
|
|
if ports.tserverPorts.ui > 0 {
|
|
servicePorts = []v1.ServicePort{
|
|
{
|
|
Name: uiPortName,
|
|
Port: ports.tserverPorts.ui,
|
|
TargetPort: intstr.FromInt(int(ports.tserverPorts.ui)),
|
|
},
|
|
}
|
|
} else {
|
|
servicePorts = nil
|
|
}
|
|
}
|
|
|
|
return servicePorts
|
|
}
|
|
|
|
func getPortsFromSpec(networkSpec yugabytedbv1alpha1.NetworkSpec) (clusterPort *clusterPorts, err error) {
|
|
ports := clusterPorts{}
|
|
|
|
for _, p := range networkSpec.Ports {
|
|
switch p.Name {
|
|
case masterUIPortName:
|
|
ports.masterPorts.ui = p.Port
|
|
case masterRPCPortName:
|
|
ports.masterPorts.rpc = p.Port
|
|
case tserverUIPortName:
|
|
ports.tserverPorts.ui = p.Port
|
|
case tserverRPCPortName:
|
|
ports.tserverPorts.rpc = p.Port
|
|
case tserverCassandraPortName:
|
|
ports.tserverPorts.cassandra = p.Port
|
|
case tserverRedisPortName:
|
|
ports.tserverPorts.redis = p.Port
|
|
case tserverPostgresPortName:
|
|
ports.tserverPorts.postgres = p.Port
|
|
default:
|
|
return &clusterPorts{}, fmt.Errorf("Invalid port name: %s. Must be one of: [%s, %s, %s, %s, %s, %s, %s]", p.Name,
|
|
masterUIPortName, masterRPCPortName, tserverUIPortName, tserverRPCPortName, tserverCassandraPortName,
|
|
tserverRedisPortName, tserverPostgresPortName)
|
|
}
|
|
}
|
|
|
|
if ports.masterPorts.ui == 0 {
|
|
ports.masterPorts.ui = masterUIPortDefault
|
|
}
|
|
|
|
if ports.masterPorts.rpc == 0 {
|
|
ports.masterPorts.rpc = masterRPCPortDefault
|
|
}
|
|
|
|
if ports.tserverPorts.rpc == 0 {
|
|
ports.tserverPorts.rpc = tserverRPCPortDefault
|
|
}
|
|
|
|
if ports.tserverPorts.cassandra == 0 {
|
|
ports.tserverPorts.cassandra = tserverCassandraPortDefault
|
|
}
|
|
|
|
if ports.tserverPorts.redis == 0 {
|
|
ports.tserverPorts.redis = tserverRedisPortDefault
|
|
}
|
|
|
|
if ports.tserverPorts.postgres == 0 {
|
|
ports.tserverPorts.postgres = tserverPostgresPortDefault
|
|
}
|
|
|
|
return &ports, nil
|
|
}
|
|
|
|
func createMasterContainerCommand(namespace, serviceName string, grpcPort, replicas int32) []string {
|
|
command := []string{
|
|
"/home/yugabyte/bin/yb-master",
|
|
fmt.Sprintf("--fs_data_dirs=%s", volumeMountPath),
|
|
fmt.Sprintf("--rpc_bind_addresses=$(POD_IP):%d", grpcPort),
|
|
fmt.Sprintf("--server_broadcast_addresses=$(POD_NAME).%s:%d", serviceName, grpcPort),
|
|
"--use_private_ip=never",
|
|
fmt.Sprintf("--master_addresses=%s.%s.svc.cluster.local:%d", serviceName, namespace, grpcPort),
|
|
"--use_initial_sys_catalog_snapshot=true",
|
|
fmt.Sprintf("--master_replication_factor=%d", replicas),
|
|
"--logtostderr",
|
|
}
|
|
return command
|
|
}
|
|
|
|
func createTServerContainerCommand(namespace, serviceName, masterServiceName string, masterGRPCPort, tserverGRPCPort, pgsqlPort, replicas int32) []string {
|
|
command := []string{
|
|
"/home/yugabyte/bin/yb-tserver",
|
|
fmt.Sprintf("--fs_data_dirs=%s", volumeMountPath),
|
|
fmt.Sprintf("--rpc_bind_addresses=$(POD_IP):%d", tserverGRPCPort),
|
|
fmt.Sprintf("--server_broadcast_addresses=$(POD_NAME).%s:%d", serviceName, tserverGRPCPort),
|
|
"--start_pgsql_proxy",
|
|
fmt.Sprintf("--pgsql_proxy_bind_address=$(POD_IP):%d", pgsqlPort),
|
|
"--use_private_ip=never",
|
|
fmt.Sprintf("--tserver_master_addrs=%s.%s.svc.cluster.local:%d", masterServiceName, namespace, masterGRPCPort),
|
|
fmt.Sprintf("--tserver_master_replication_factor=%d", replicas),
|
|
"--logtostderr",
|
|
}
|
|
return command
|
|
}
|
|
|
|
func createMasterContainerPortsList(clusterPortsSpec *clusterPorts) []v1.ContainerPort {
|
|
ports := []v1.ContainerPort{
|
|
{
|
|
Name: masterContainerUIPortName,
|
|
ContainerPort: int32(clusterPortsSpec.masterPorts.ui),
|
|
},
|
|
{
|
|
Name: masterContainerRPCPortName,
|
|
ContainerPort: int32(clusterPortsSpec.masterPorts.rpc),
|
|
},
|
|
}
|
|
|
|
return ports
|
|
}
|
|
|
|
func createTServerContainerPortsList(clusterPortsSpec *clusterPorts) []v1.ContainerPort {
|
|
tserverUIPort := int32(clusterPortsSpec.tserverPorts.ui)
|
|
|
|
if tserverUIPort <= 0 {
|
|
tserverUIPort = tserverUIPortDefault
|
|
}
|
|
|
|
ports := []v1.ContainerPort{
|
|
{
|
|
Name: tserverContainerUIPortName,
|
|
ContainerPort: tserverUIPort,
|
|
},
|
|
{
|
|
Name: tserverContainerRPCPortName,
|
|
ContainerPort: int32(clusterPortsSpec.tserverPorts.rpc),
|
|
},
|
|
{
|
|
Name: cassandraPortName,
|
|
ContainerPort: int32(clusterPortsSpec.tserverPorts.cassandra),
|
|
},
|
|
{
|
|
Name: redisPortName,
|
|
ContainerPort: int32(clusterPortsSpec.tserverPorts.redis),
|
|
},
|
|
{
|
|
Name: postgresPortName,
|
|
ContainerPort: int32(clusterPortsSpec.tserverPorts.postgres),
|
|
},
|
|
}
|
|
|
|
return ports
|
|
}
|
|
|
|
func (c *cluster) addCRNameSuffix(str string) string {
|
|
return fmt.Sprintf("%s-%s", str, c.name)
|
|
}
|