repair json decode problem

cmd/mysql-agent/app/mysql_agent.go

@@ -22,6 +22,8 @@ import (
 	"strconv"
 	"sync"
 	"time"
+	"os"
+	"fmt"
 
 	"github.com/golang/glog"
 	"github.com/heptiolabs/healthcheck"
@@ -100,15 +102,17 @@ func Run(opts *agentopts.MySQLAgentOpts) error {
 	if err != nil {
 		return errors.Wrap(err, "failed to create new local MySQL InnoDB cluster manager")
 	}
-
-	// Initialise the agent metrics.
+	// agent prometheus port
+    agentPromePort := os.Getenv("AGENT_PROME_PORT")
+    promeMetricsEndpoint := fmt.Sprintf("0.0.0.0: %s", agentPromePort)
+    glog.Info("agent prometheus endpoint: ", promeMetricsEndpoint)
 	metrics.RegisterPodName(opts.Hostname)
 	metrics.RegisterClusterName(manager.Instance.ClusterName)
 	clustermgr.RegisterMetrics()
 	backupcontroller.RegisterMetrics()
 	restorecontroller.RegisterMetrics()
 	http.Handle("/metrics", prometheus.Handler())
-	go http.ListenAndServe(metricsEndpoint, nil)
+	go http.ListenAndServe(promeMetricsEndpoint, nil)
 
 	// Block until local instance successfully initialised.
 	for !manager.Sync(ctx) {

pkg/apis/mysql/v1alpha1/helpers.go

@@ -22,8 +22,13 @@ import (
 const (
 	// DefaultVersion is the MySQL version to use if not specified explicitly by user
 	DefaultVersion      = "8.0.12"
+	DefaultReplicationGroupPort = 33061
+	DefaultAgentHealthCheckPort = 10512
+	DefaultAgentPromePort = 8080
+	DefaultMysqlPort = 3306
 	defaultMembers      = 3
 	defaultBaseServerID = 1000
+	defaultAgentIntervalTime = 15
 	// maxBaseServerID is the maximum safe value for BaseServerID calculated
 	// as max MySQL server_id value - max Replication Group size.
 	maxBaseServerID uint32 = 4294967295 - 9
@@ -69,6 +74,24 @@ func (c *Cluster) EnsureDefaults() *Cluster {
 		c.Spec.Version = DefaultVersion
 	}
 
+	if c.Spec.GroupPort == 0 {
+		c.Spec.GroupPort = DefaultReplicationGroupPort
+	}
+
+	if c.Spec.AgentCheckPort == 0 {
+		c.Spec.AgentCheckPort = DefaultAgentHealthCheckPort
+	}
+
+	if c.Spec.AgentPromePort == 0 {
+		c.Spec.AgentPromePort = DefaultAgentPromePort
+	}
+
+	if c.Spec.MysqlPort == 0 {
+		c.Spec.MysqlPort = DefaultMysqlPort
+	}
+	if c.Spec.AgentIntervalTime == 0 {
+		c.Spec.AgentIntervalTime = defaultAgentIntervalTime
+	}
 	return c
 }
 

pkg/apis/mysql/v1alpha1/types.go

@@ -43,6 +43,18 @@ type ClusterSpec struct {
 	// all instances will be R/W. If false (the default), only a single instance
 	// will be R/W and the rest will be R/O.
 	MultiMaster bool `json:"multiMaster,omitempty"`
+	//hostnetwork
+	HostNetwork bool `json:"hostNetwork,omitempty"`
+	//replicationGroupPort 
+	GroupPort uint32 `json:"groupPort,omitempty"`
+	//agent healthcheck port
+	AgentCheckPort uint32 `json:"agentCheckPort,omitempty"`
+	//agent prometheus port
+	AgentPromePort uint32 `json:"agentPromePort,omitempty"`
+	//mysql_port
+	MysqlPort uint32 `json:"mysqlPort,omitempty"`
+	//agent execute interval
+	AgentIntervalTime uint32 `json:"agentIntervalTime,omitempty"`
 	// NodeSelector is a selector which must be true for the pod to fit on a node.
 	// Selector which must match a node's labels for the pod to be scheduled on that node.
 	// More info: https://kubernetes.io/docs/concepts/configuration/assign-pod-node/

pkg/cluster/innodb/innodb.go

@@ -17,6 +17,7 @@ package innodb
 import (
 	"fmt"
 	"net"
+	"os"
 )
 
 // DefaultClusterName is the default name assigned to InnoDB clusters created by
@@ -123,7 +124,7 @@ func (s *ClusterStatus) GetInstanceStatus(name string) InstanceStatus {
 	if s.DefaultReplicaSet.Topology == nil {
 		return InstanceStatusNotFound
 	}
-	if is, ok := s.DefaultReplicaSet.Topology[fmt.Sprintf("%s:%d", name, MySQLDBPort)]; ok {
+	if is, ok := s.DefaultReplicaSet.Topology[fmt.Sprintf("%s:%s", name, os.Getenv("MYSQL_PORT"))]; ok {
 		return is.Status
 	}
 	return InstanceStatusNotFound

pkg/cluster/instance.go

@@ -24,7 +24,7 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/oracle/mysql-operator/pkg/cluster/innodb"
+	//"github.com/oracle/mysql-operator/pkg/cluster/innodb"
 )
 
 // Instance represents the local MySQL instance.
@@ -62,18 +62,19 @@ func NewInstance(namespace, clusterName, parentName string, ordinal, port int, m
 // NewLocalInstance creates a new instance of this structure, with it's name and index
 // populated from os.Hostname().
 func NewLocalInstance() (*Instance, error) {
-	hostname, err := os.Hostname()
-	if err != nil {
-		return nil, err
+	pod_name := os.Getenv("MY_POD_NAME")
+	if pod_name == "" {
+		return nil, errors.Errorf("env MY_POD_NAME is empty!!!")
 	}
-	name, ordinal := GetParentNameAndOrdinal(hostname)
+	name, ordinal := GetParentNameAndOrdinal(pod_name)
 	multiMaster, _ := strconv.ParseBool(os.Getenv("MYSQL_CLUSTER_MULTI_MASTER"))
+	mysqlPort, _ := strconv.ParseInt(os.Getenv("MYSQL_PORT"), 10, 32)
 	return &Instance{
 		Namespace:   os.Getenv("POD_NAMESPACE"),
 		ClusterName: os.Getenv("MYSQL_CLUSTER_NAME"),
 		ParentName:  name,
 		Ordinal:     ordinal,
-		Port:        innodb.MySQLDBPort,
+		Port:        int(mysqlPort),
 		MultiMaster: multiMaster,
 		IP:          net.ParseIP(os.Getenv("MY_POD_IP")),
 	}, nil
@@ -90,12 +91,13 @@ func NewInstanceFromGroupSeed(seed string) (*Instance, error) {
 	// MySQLDB port not its group replication port.
 	parentName, ordinal := GetParentNameAndOrdinal(podName)
 	multiMaster, _ := strconv.ParseBool(os.Getenv("MYSQL_CLUSTER_MULTI_MASTER"))
+	mysqlPort, _ := strconv.ParseInt(os.Getenv("MYSQL_PORT"), 10, 32)
 	return &Instance{
 		ClusterName: os.Getenv("MYSQL_CLUSTER_NAME"),
 		Namespace:   os.Getenv("POD_NAMESPACE"),
 		ParentName:  parentName,
 		Ordinal:     ordinal,
-		Port:        innodb.MySQLDBPort,
+		Port:        int(mysqlPort),
 		MultiMaster: multiMaster,
 	}, nil
 }

pkg/controllers/cluster/manager/cluster_manager.go

@@ -17,7 +17,9 @@ package manager
 import (
 	"context"
 	"fmt"
+	"os"
 	"strings"
+	"strconv"
 	"time"
 
 	"github.com/golang/glog"
@@ -36,7 +38,7 @@ import (
 	"github.com/oracle/mysql-operator/pkg/util/mysqlsh"
 )
 
-const pollingIntervalSeconds = 15
+const pollingIntervalSeconds = 60
 
 // ClusterManager manages the local MySQL instance's membership of an InnoDB cluster.
 type ClusterManager struct {
@@ -100,6 +102,7 @@ func (m *ClusterManager) getClusterStatus(ctx context.Context) (*innodb.ClusterS
 	if localMSHErr != nil {
 		var err error
 		clusterStatus, err = getClusterStatusFromGroupSeeds(ctx, m.kubeClient, m.Instance)
+		glog.V(2).Infof("get cluster seeds*** error: %+v", err)
 		if err != nil {
 			// NOTE: We return the localMSHErr rather than the error here so that we
 			// can dispatch on it.
@@ -122,6 +125,7 @@ func (m *ClusterManager) Sync(ctx context.Context) bool {
 	clusterStatus, err := m.getClusterStatus(ctx)
 	if err != nil {
 		myshErr, ok := errors.Cause(err).(*mysqlsh.Error)
+		glog.V(2).Infof("get cluster*** error: %+v", err)
 		if !ok {
 			glog.Errorf("Failed to get the cluster status: %+v", err)
 			return false
@@ -289,10 +293,17 @@ func (m *ClusterManager) handleInstanceNotFound(ctx context.Context, primaryAddr
 		glog.Errorf("Getting CIDR to whitelist for GR: %v", err)
 		return false
 	}
+	//use deault mysqlPort + 1
+	//localAddress := fmt.Sprintf("%s:%s", m.Instance.Name(), os.Getenv("GROUP_PORT"))
+    //groupSeeds := os.Getenv("REPLICATION_GROUP_SEEDS")
+
+    //glog.Infof("localAddress: %s, groupSeeds: %s", localAddress, groupSeeds)
 
 	if err := psh.AddInstanceToCluster(ctx, m.Instance.GetShellURI(), mysqlsh.Options{
 		"memberSslMode": "REQUIRED",
 		"ipWhitelist":   whitelistCIDR,
+	//	"localAddress": localAddress,
+	//	"groupSeeds": groupSeeds,
 	}); err != nil {
 		glog.Errorf("Failed to add to cluster: %v", err)
 		return false
@@ -322,9 +333,18 @@ func (m *ClusterManager) createCluster(ctx context.Context) (*innodb.ClusterStat
 	if err != nil {
 		return nil, errors.Wrap(err, "getting CIDR to whitelist for  GR")
 	}
+
+	// use deault mysql_port + 1
+	//localAddress := fmt.Sprintf("%s:%s", m.Instance.Name(), os.Getenv("GROUP_PORT"))
+    //groupSeeds := os.Getenv("REPLICATION_GROUP_SEEDS")
+
+    //glog.Infof("localAddress: %s, groupSeeds: %s", localAddress, groupSeeds)
+
 	opts := mysqlsh.Options{
 		"memberSslMode": "REQUIRED",
 		"ipWhitelist":   whitelistCIDR,
+	//	"localAddress": localAddress,
+	//	"groupSeeds": groupSeeds,
 	}
 	if m.Instance.MultiMaster {
 		opts["force"] = "True"
@@ -355,7 +375,10 @@ func (m *ClusterManager) rebootFromOutage(ctx context.Context) (*innodb.ClusterS
 // Run runs the ClusterManager controller.
 // NOTE: ctx is not currently used for cancellation by caller (the stopCh is).
 func (m *ClusterManager) Run(ctx context.Context) {
-	wait.Until(func() { m.Sync(ctx) }, time.Second*pollingIntervalSeconds, ctx.Done())
+	interval_time, _ := strconv.ParseUint(os.Getenv("AGENT_INTERVAL"), 10, 32)
+	glog.Info("***agent run interval: ", interval_time)
+	//wait.Until(func() { m.Sync(ctx) }, time.Second*pollingIntervalSeconds, ctx.Done())
+	wait.Until(func() { m.Sync(ctx) }, time.Second*time.Duration(interval_time), ctx.Done())
 
 	<-ctx.Done()
 

pkg/controllers/cluster/manager/innodb_cluster.go

@@ -35,7 +35,7 @@ import (
 
 var errNoClusterFound = errors.New("no cluster found on any of the seed nodes")
 
-const defaultTimeout = 10 * time.Second
+const defaultTimeout = 60 * time.Second
 
 // isDatabaseRunning returns true if a connection can be made to the MySQL
 // database running in the pod instance in which this function is called.
@@ -46,6 +46,7 @@ func isDatabaseRunning(ctx context.Context) bool {
 		"mysqladmin",
 		"--protocol", "tcp",
 		"-u", "root",
+		os.ExpandEnv("-P$MYSQL_PORT"),
 		os.ExpandEnv("-p$MYSQL_ROOT_PASSWORD"),
 		"status",
 	).Run()

pkg/options/agent/options.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"os"
 	"time"
+	"strconv"
 
 	"github.com/golang/glog"
 	"github.com/spf13/pflag"
@@ -65,8 +66,10 @@ func NewMySQLAgentOpts() *MySQLAgentOpts {
 	}
 	namespace := os.Getenv("POD_NAMESPACE")
 	clusterName := os.Getenv("MYSQL_CLUSTER_NAME")
+	healthcheckPort, _ := strconv.ParseInt(os.Getenv("AGENT_HEALTHCHECK_PORT"), 10, 32)
+	glog.V(2).Infof("mysql-agent healthcheckPort: %d", healthcheckPort)
 	return &MySQLAgentOpts{
-		HealthcheckPort: DefaultMySQLAgentHeathcheckPort,
+		HealthcheckPort: int32(healthcheckPort),
 		Address:         "0.0.0.0",
 		Namespace:       namespace,
 		ClusterName:     clusterName,

pkg/resources/services/service.go

@@ -25,7 +25,7 @@ import (
 
 // NewForCluster will return a new headless Kubernetes service for a MySQL cluster
 func NewForCluster(cluster *v1alpha1.Cluster) *corev1.Service {
-	mysqlPort := corev1.ServicePort{Port: 3306}
+	mysqlPort := corev1.ServicePort{Port: int32(cluster.Spec.MysqlPort)}
 	svc := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Labels:    map[string]string{constants.ClusterLabel: cluster.Name},