Reporting for multi-node EC installations

Author: sgalsaleh

pkg/api/reporting/types/types.go

@@ -15,6 +15,7 @@ type ReportingInfo struct {
 	KURLInstallID           string         `json:"kurl_install_id" yaml:"kurl_install_id"`
 	EmbeddedClusterID       string         `json:"embedded_cluster_id" yaml:"embedded_cluster_id"`
 	EmbeddedClusterVersion  string         `json:"embedded_cluster_version" yaml:"embedded_cluster_version"`
+	EmbeddedClusterNodes    string         `json:"embedded_cluster_nodes" yaml:"embedded_cluster_nodes"`
 	IsGitOpsEnabled         bool           `json:"is_gitops_enabled" yaml:"is_gitops_enabled"`
 	GitOpsProvider          string         `json:"gitops_provider" yaml:"gitops_provider"`
 	SnapshotProvider        string         `json:"snapshot_provider" yaml:"snapshot_provider"`

pkg/apiserver/server.go

@@ -15,7 +15,6 @@ import (
 	"github.com/replicatedhq/kots/pkg/binaries"
 	"github.com/replicatedhq/kots/pkg/handlers"
 	identitymigrate "github.com/replicatedhq/kots/pkg/identity/migrate"
-	"github.com/replicatedhq/kots/pkg/informers"
 	"github.com/replicatedhq/kots/pkg/k8sutil"
 	"github.com/replicatedhq/kots/pkg/operator"
 	operatorclient "github.com/replicatedhq/kots/pkg/operator/client"
@@ -31,6 +30,7 @@ import (
 	"github.com/replicatedhq/kots/pkg/updatechecker"
 	"github.com/replicatedhq/kots/pkg/upgradeservice"
 	"github.com/replicatedhq/kots/pkg/util"
+	"github.com/replicatedhq/kots/pkg/watchers"
 	"golang.org/x/crypto/bcrypt"
 )
 
@@ -121,8 +121,8 @@ func Start(params *APIServerParams) {
 
 	supportbundle.StartServer()
 
-	if err := informers.Start(); err != nil {
-		log.Println("Failed to start informers:", err)
+	if err := watchers.Start(op.GetClusterID()); err != nil {
+		log.Println("Failed to start watchers:", err)
 	}
 
 	if err := updatechecker.Start(); err != nil {

pkg/embeddedcluster/helmvm_node.go renamed to pkg/embeddedcluster/node.go

@@ -126,6 +126,10 @@ func startLoop(fn func(), intervalInSeconds time.Duration) {
 	}()
 }
 
+func (o *Operator) GetClusterID() string {
+	return o.clusterID
+}
+
 func (o *Operator) resumeDeployments() {
 	apps, err := o.store.ListAppsForDownstream(o.clusterID)
 	if err != nil {

pkg/embeddedcluster/helmvm_nodes.go renamed to pkg/embeddedcluster/nodes.go

@@ -11,6 +11,7 @@ import (
 	downstreamtypes "github.com/replicatedhq/kots/pkg/api/downstream/types"
 	"github.com/replicatedhq/kots/pkg/api/reporting/types"
 	"github.com/replicatedhq/kots/pkg/buildversion"
+	"github.com/replicatedhq/kots/pkg/embeddedcluster"
 	"github.com/replicatedhq/kots/pkg/gitops"
 	"github.com/replicatedhq/kots/pkg/k8sutil"
 	"github.com/replicatedhq/kots/pkg/kotsadm"
@@ -157,6 +158,7 @@ func GetReportingInfo(appID string) *types.ReportingInfo {
 		r.AppStatus = string(appStatus.State)
 	}
 
+	// kurl
 	r.IsKurl, err = kurl.IsKurl(clientset)
 	if err != nil {
 		logger.Debugf(errors.Wrap(err, "failed to check if cluster is kurl").Error())
@@ -167,11 +169,28 @@ func GetReportingInfo(appID string) *types.ReportingInfo {
 		if err != nil {
 			logger.Debugf(errors.Wrap(err, "failed to get kurl nodes").Error())
 		}
+		if kurlNodes != nil {
+			for _, kurlNode := range kurlNodes.Nodes {
+				r.KurlNodeCountTotal++
+				if kurlNode.IsConnected && kurlNode.IsReady {
+					r.KurlNodeCountReady++
+				}
+			}
+		}
+	}
 
-		for _, kurlNode := range kurlNodes.Nodes {
-			r.KurlNodeCountTotal++
-			if kurlNode.IsConnected && kurlNode.IsReady {
-				r.KurlNodeCountReady++
+	// embedded cluster
+	if util.IsEmbeddedCluster() && clientset != nil {
+		ecNodes, err := embeddedcluster.GetNodes(context.TODO(), clientset)
+		if err != nil {
+			logger.Debugf("failed to get embedded cluster nodes: %v", err.Error())
+		}
+		if ecNodes != nil {
+			marshalled, err := json.Marshal(ecNodes.Nodes)
+			if err != nil {
+				logger.Debugf("failed to marshal embedded cluster node: %v", err.Error())
+			} else {
+				r.EmbeddedClusterNodes = string(marshalled)
 			}
 		}
 	}

pkg/informers/informers.go

@@ -69,6 +69,7 @@ func BuildInstanceReport(licenseID string, reportingInfo *types.ReportingInfo) *
 				KurlInstallID:             reportingInfo.KURLInstallID,
 				EmbeddedClusterID:         reportingInfo.EmbeddedClusterID,
 				EmbeddedClusterVersion:    reportingInfo.EmbeddedClusterVersion,
+				EmbeddedClusterNodes:      reportingInfo.EmbeddedClusterNodes,
 				IsGitOpsEnabled:           reportingInfo.IsGitOpsEnabled,
 				GitOpsProvider:            reportingInfo.GitOpsProvider,
 				SnapshotProvider:          reportingInfo.SnapshotProvider,

pkg/operator/operator.go

@@ -29,6 +29,7 @@ type InstanceReportEvent struct {
 	KurlInstallID             string `json:"kurl_install_id,omitempty"`
 	EmbeddedClusterID         string `json:"embedded_cluster_id,omitempty"`
 	EmbeddedClusterVersion    string `json:"embedded_cluster_version,omitempty"`
+	EmbeddedClusterNodes      string `json:"embedded_cluster_nodes"`
 	IsGitOpsEnabled           bool   `json:"is_gitops_enabled"`
 	GitOpsProvider            string `json:"gitops_provider"`
 	SnapshotProvider          string `json:"snapshot_provider"`

pkg/reporting/app.go

@@ -353,6 +353,7 @@ func createTestInstanceEvent(reportedAt int64) InstanceReportEvent {
 		KurlInstallID:             "test-kurl-install-id",
 		EmbeddedClusterID:         "test-embedded-cluster-id",
 		EmbeddedClusterVersion:    "test-embedded-cluster-version",
+		EmbeddedClusterNodes:      `[{"name":"node-1","isReady":true},{"name":"node-2","isReady":false},{"name":"node-3","isReady":true},{"name":"node-4","isReady":false}]`,
 		IsGitOpsEnabled:           true,
 		GitOpsProvider:            "test-gitops-provider",
 		SnapshotProvider:          "test-snapshot-provider",

pkg/reporting/app_airgap.go

@@ -68,6 +68,9 @@ func GetReportingInfoHeaders(reportingInfo *types.ReportingInfo) map[string]stri
 	if reportingInfo.EmbeddedClusterVersion != "" {
 		headers["X-Replicated-EmbeddedClusterVersion"] = reportingInfo.EmbeddedClusterVersion
 	}
+	if reportingInfo.EmbeddedClusterNodes != "" {
+		headers["X-Replicated-EmbeddedClusterNodes"] = reportingInfo.EmbeddedClusterNodes
+	}
 
 	headers["X-Replicated-KurlNodeCountTotal"] = strconv.Itoa(reportingInfo.KurlNodeCountTotal)
 	headers["X-Replicated-KurlNodeCountReady"] = strconv.Itoa(reportingInfo.KurlNodeCountReady)

pkg/reporting/instance_report.go

@@ -0,0 +1,119 @@
+package watchers
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"github.com/replicatedhq/kots/pkg/k8sutil"
+	"github.com/replicatedhq/kots/pkg/logger"
+	"github.com/replicatedhq/kots/pkg/reporting"
+	"github.com/replicatedhq/kots/pkg/store"
+	"github.com/replicatedhq/kots/pkg/util"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+func Start(clusterID string) error {
+	cfg, err := k8sutil.GetClusterConfig()
+	if err != nil {
+		return errors.Wrap(err, "get cluster config")
+	}
+
+	clientset, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return errors.Wrap(err, "create clientset")
+	}
+
+	if err := watchECNodes(clientset, clusterID); err != nil {
+		return errors.Wrap(err, "watch embedded cluster nodes")
+	}
+
+	return nil
+}
+
+func watchECNodes(clientset kubernetes.Interface, clusterID string) error {
+	if !util.IsEmbeddedCluster() {
+		return nil
+	}
+
+	logger.Info("starting embedded cluster nodes watcher")
+
+	factory := informers.NewSharedInformerFactory(clientset, 0)
+	nodeInformer := factory.Core().V1().Nodes().Informer()
+
+	// by default, add func gets called for existing nodes, and
+	// we don't want to report N times every time kotsadm restarts,
+	// specially if there are a lot of nodes.
+	hasSynced := false
+
+	nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			if !hasSynced {
+				return
+			}
+			node := obj.(*corev1.Node)
+			logger.Infof("Node added: %s", node.Name)
+			if err := submitAppInfo(clusterID); err != nil {
+				logger.Debugf("failed to submit app info: %v", err)
+			}
+		},
+		DeleteFunc: func(obj interface{}) {
+			node := obj.(*corev1.Node)
+			logger.Infof("Node deleted: %s", node.Name)
+			if err := submitAppInfo(clusterID); err != nil {
+				logger.Debugf("failed to submit app info: %v", err)
+			}
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			oldNode := oldObj.(*corev1.Node)
+			newNode := newObj.(*corev1.Node)
+
+			// Check if ready condition changed
+			oldReady := getNodeReadyStatus(oldNode)
+			newReady := getNodeReadyStatus(newNode)
+
+			if oldReady != newReady {
+				logger.Infof("Node %s ready status changed from %v to %v", newNode.Name, oldReady, newReady)
+				if err := submitAppInfo(clusterID); err != nil {
+					logger.Debugf("failed to submit app info: %v", err)
+				}
+			}
+		},
+	})
+
+	ctx := context.Background()
+	go nodeInformer.Run(ctx.Done())
+
+	if !cache.WaitForCacheSync(ctx.Done(), nodeInformer.HasSynced) {
+		return errors.New("sync node cache")
+	}
+
+	hasSynced = true
+
+	return nil
+}
+
+func getNodeReadyStatus(node *corev1.Node) corev1.ConditionStatus {
+	for _, condition := range node.Status.Conditions {
+		if condition.Type == corev1.NodeReady {
+			return condition.Status
+		}
+	}
+	return corev1.ConditionUnknown
+}
+
+func submitAppInfo(clusterID string) error {
+	apps, err := store.GetStore().ListAppsForDownstream(clusterID)
+	if err != nil {
+		return errors.Wrap(err, "list installed apps for downstream")
+	}
+
+	if len(apps) == 0 {
+		return nil
+	}
+
+	// embedded cluster only supports one app
+	return reporting.GetReporter().SubmitAppInfo(apps[0].ID)
+}