Reporting for multi-node EC installations

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 → pkg/embeddedcluster/node.go

@@ -131,7 +131,7 @@ func nodeMetrics(ctx context.Context, client kubernetes.Interface, metricsClient
 		CPU:              cpuCapacity,
 		Memory:           memoryCapacity,
 		Pods:             podCapacity,
-		Labels:           nodeRolesFromLabels(node.Labels),
+		Roles:            nodeRolesFromLabels(node.Labels),
 		Conditions:       findNodeConditions(node.Status.Conditions),
 		PodList:          podInfo,
 	}, nil

pkg/embeddedcluster/types/types.go

@@ -22,7 +22,7 @@ type Node struct {
 	CPU              CapacityUsed   `json:"cpu"`
 	Memory           CapacityUsed   `json:"memory"`
 	Pods             CapacityUsed   `json:"pods"`
-	Labels           []string       `json:"labels"`
+	Roles            []string       `json:"roles"`
 	Conditions       NodeConditions `json:"conditions"`
 	PodList          []PodInfo      `json:"podList"`
 }

pkg/operator/operator.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/reporting/app.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"
@@ -120,18 +121,18 @@ func GetReportingInfo(appID string) *types.ReportingInfo {
 
 	cfg, err := k8sutil.GetClusterConfig()
 	if err != nil {
-		logger.Debugf("failed to get cluster config: %v", err.Error())
+		logger.Warnf("failed to get cluster config: %v", err.Error())
 	}
 
 	clientset, err := kubernetes.NewForConfig(cfg)
 	if err != nil {
-		logger.Debugf("failed to get clientset: %v", err.Error())
+		logger.Warnf("failed to get clientset: %v", err.Error())
 	}
 	r.ClusterID = k8sutil.GetKotsadmID(clientset)
 
 	di, err := getDownstreamInfo(appID)
 	if err != nil {
-		logger.Debugf("failed to get downstream info: %v", err.Error())
+		logger.Warnf("failed to get downstream info: %v", err.Error())
 	}
 	if di != nil {
 		r.Downstream = *di
@@ -140,7 +141,7 @@ func GetReportingInfo(appID string) *types.ReportingInfo {
 	// get kubernetes cluster version
 	k8sVersion, err := k8sutil.GetK8sVersion(clientset)
 	if err != nil {
-		logger.Debugf("failed to get k8s version: %v", err.Error())
+		logger.Warnf("failed to get k8s version: %v", err.Error())
 	} else {
 		r.K8sVersion = k8sVersion
 	}
@@ -152,26 +153,44 @@ func GetReportingInfo(appID string) *types.ReportingInfo {
 	// get app status
 	appStatus, err := store.GetStore().GetAppStatus(appID)
 	if err != nil {
-		logger.Debugf("failed to get app status: %v", err.Error())
+		logger.Warnf("failed to get app status: %v", err.Error())
 	} else {
 		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())
+		logger.Warnf(errors.Wrap(err, "failed to check if cluster is kurl").Error())
 	}
 
 	if r.IsKurl && clientset != nil {
 		kurlNodes, err := kurl.GetNodes(clientset)
 		if err != nil {
-			logger.Debugf(errors.Wrap(err, "failed to get kurl nodes").Error())
+			logger.Warnf(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.Warnf("failed to get embedded cluster nodes: %v", err.Error())
+		}
+		if ecNodes != nil {
+			marshalled, err := json.Marshal(ecNodes.Nodes)
+			if err != nil {
+				logger.Warnf("failed to marshal embedded cluster node: %v", err.Error())
+			} else {
+				r.EmbeddedClusterNodes = string(marshalled)
 			}
 		}
 	}
@@ -180,17 +199,17 @@ func GetReportingInfo(appID string) *types.ReportingInfo {
 
 	veleroClient, err := k8sutil.GetKubeClient(context.TODO())
 	if err != nil {
-		logger.Debugf("failed to get velero client: %v", err.Error())
+		logger.Warnf("failed to get velero client: %v", err.Error())
 	}
 
 	if clientset != nil && veleroClient != nil {
 		bsl, err := snapshot.FindBackupStoreLocation(context.TODO(), clientset, veleroClient, util.PodNamespace)
 		if err != nil {
-			logger.Debugf("failed to find backup store location: %v", err.Error())
+			logger.Warnf("failed to find backup store location: %v", err.Error())
 		} else {
 			report, err := getSnapshotReport(store.GetStore(), bsl, appID, r.ClusterID)
 			if err != nil {
-				logger.Debugf("failed to get snapshot report: %v", err.Error())
+				logger.Warnf("failed to get snapshot report: %v", err.Error())
 			} else {
 				r.SnapshotProvider = report.Provider
 				r.SnapshotFullSchedule = report.FullSchedule
@@ -212,7 +231,9 @@ func getDownstreamInfo(appID string) (*types.DownstreamInfo, error) {
 		return nil, errors.Wrap(err, "failed to list downstreams for app")
 	}
 	if len(downstreams) == 0 {
-		return nil, errors.New("no downstreams found for app")
+		// this can happen during initial install workflow until the app is installed and has downstreams
+		logger.Debugf("no downstreams found for app")
+		return nil, nil
 	}
 
 	currentVersion, err := store.GetStore().GetCurrentDownstreamVersion(appID, downstreams[0].ClusterID)
@@ -247,7 +268,7 @@ func getDownstreamInfo(appID string) (*types.DownstreamInfo, error) {
 
 		var preflightResults *troubleshootpreflight.UploadPreflightResults
 		if err := json.Unmarshal([]byte(currentVersion.PreflightResult), &preflightResults); err != nil {
-			logger.Debugf("failed to unmarshal preflight results: %v", err.Error())
+			logger.Warnf("failed to unmarshal preflight results: %v", err.Error())
 		}
 		di.PreflightState = getPreflightState(preflightResults)
 		di.SkipPreflights = currentVersion.PreflightSkipped
@@ -269,7 +290,7 @@ func getDownstreamInfo(appID string) (*types.DownstreamInfo, error) {
 func getGitOpsReport(clientset kubernetes.Interface, appID string, clusterID string) (bool, string) {
 	gitOpsConfig, err := gitops.GetDownstreamGitOpsConfig(clientset, appID, clusterID)
 	if err != nil {
-		logger.Debugf("failed to get gitops config: %v", err.Error())
+		logger.Warnf("failed to get gitops config: %v", err.Error())
 		return false, ""
 	}
 

pkg/reporting/app_airgap.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/reporting/instance_report.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/report_test.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/util.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)