Merge pull request #683 from rancher/logging-status

Add Logging Status API

go.sum

@@ -1887,8 +1887,6 @@ github.com/kralicky/ragu v1.0.0-rc3 h1:auI0G5WcLcBC18gQABjpykvEbzEt2xStQsC6dxuva
 github.com/kralicky/ragu v1.0.0-rc3/go.mod h1:upcZm+aSNq1ngZDshp85T4tDG8rhfO9Ak66sY/4bI2g=
 github.com/kralicky/spellbook v0.0.0-20220829172922-3d415e02ee8a h1:RfpqOGzhMSF0OWvI8AqYxlvhuVZ72EQiYu9ClM89bKg=
 github.com/kralicky/spellbook v0.0.0-20220829172922-3d415e02ee8a/go.mod h1:qp6R4XKeuJk2j8nHke4Y9rBEpc2HkCjXR/gssL3OpOU=
-github.com/kralicky/totem v1.1.0-rc6.0.20221015041849-acb5d4c633f2 h1:eEMfmOPb9Y37ZSar9KR7nzMhSC4mqbUdFSQVB4F308c=
-github.com/kralicky/totem v1.1.0-rc6.0.20221015041849-acb5d4c633f2/go.mod h1:l6wZ22m0hEIBML+HE0qJTwqAZMg9m9qpbRC2tCZNtqg=
 github.com/kralicky/totem v1.1.0-rc7 h1:/ubi2n0mcM/+1zbok8Td8GXh2W6mlEQb9lbdsHvwjJ4=
 github.com/kralicky/totem v1.1.0-rc7/go.mod h1:l6wZ22m0hEIBML+HE0qJTwqAZMg9m9qpbRC2tCZNtqg=
 github.com/kralicky/yaml/v3 v3.0.0-20220520012407-b0e7050bd81d h1:kLfaaFdmCHKZCvL4DzQ7T9YsAVSBqZez34zaegldkls=

plugins/logging/pkg/apis/loggingadmin/loggingadmin.proto

@@ -40,6 +40,11 @@ service LoggingAdmin {
             get: "/logging/storageclasses"
         };
     }
+    rpc GetOpensearchStatus(google.protobuf.Empty) returns(StatusResponse) {
+        option (google.api.http) = {
+            get: "/logging/status"
+        };
+    }
 }
 
 message OpensearchCluster {
@@ -95,4 +100,9 @@ message ResourceRequirements {
 message ComputeResourceQuantities {
     string CPU = 1;
     string Memory = 2;
+}
+
+message StatusResponse {
+    int32 status = 1;
+    string details = 2;
 }

plugins/logging/pkg/gateway/admin.go

@@ -11,14 +11,16 @@ import (
 	"time"
 
 	"github.com/lestrrat-go/backoff/v2"
-	"github.com/opensearch-project/opensearch-go"
 	osclient "github.com/opensearch-project/opensearch-go"
 	opnicorev1beta1 "github.com/rancher/opni/apis/core/v1beta1"
 	loggingv1beta1 "github.com/rancher/opni/apis/logging/v1beta1"
 	"github.com/rancher/opni/pkg/util"
 	"github.com/rancher/opni/plugins/logging/pkg/apis/loggingadmin"
 	"github.com/rancher/opni/plugins/logging/pkg/errors"
+	"github.com/rancher/opni/plugins/logging/pkg/opensearchdata"
 	"github.com/samber/lo"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/emptypb"
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
@@ -41,6 +43,33 @@ const (
 	defaultRepo       = "docker.io/rancher"
 )
 
+type ClusterStatus int
+
+const (
+	ClusterStatusPending ClusterStatus = iota + 1
+	ClusterStatusGreen
+	ClusterStatusYellow
+	ClusterStatusRed
+	ClusterStatusError
+)
+
+func ClusterStatusDescription(s ClusterStatus) string {
+	switch s {
+	case ClusterStatusPending:
+		return "Opensearch cluster is initializing"
+	case ClusterStatusGreen:
+		return "Opensearch cluster is green"
+	case ClusterStatusYellow:
+		return "Opensearch cluster is yellow"
+	case ClusterStatusRed:
+		return "Opensearch cluster is red"
+	case ClusterStatusError:
+		return "Error fetching status from Opensearch cluster"
+	default:
+		return "unknown status"
+	}
+}
+
 func (p *Plugin) GetOpensearchCluster(
 	ctx context.Context,
 	empty *emptypb.Empty,
@@ -89,7 +118,7 @@ func (p *Plugin) DeleteOpensearchCluster(
 	empty *emptypb.Empty,
 ) (*emptypb.Empty, error) {
 	// Check that it is safe to delete the cluster
-	p.opensearchClient.UnsetClient()
+	p.opensearchManager.UnsetClient()
 
 	loggingClusters := &opnicorev1beta1.LoggingClusterList{}
 	err := p.k8sClient.List(p.ctx, loggingClusters, client.InNamespace(p.storageNamespace))
@@ -121,7 +150,7 @@ func (p *Plugin) CreateOrUpdateOpensearchCluster(
 	}
 	k8sOpensearchCluster := &loggingv1beta1.OpniOpensearch{}
 
-	go p.opensearchClient.SetClient(p.setOpensearchClient)
+	go p.opensearchManager.SetClient(p.setOpensearchClient)
 	exists := true
 	err := p.k8sClient.Get(ctx, types.NamespacedName{
 		Name:      p.opensearchCluster.Name,
@@ -301,6 +330,54 @@ func (p *Plugin) GetStorageClasses(ctx context.Context, in *emptypb.Empty) (*log
 	}, nil
 }
 
+func (p *Plugin) GetOpensearchStatus(ctx context.Context, in *emptypb.Empty) (*loggingadmin.StatusResponse, error) {
+	if err := p.k8sClient.Get(ctx, types.NamespacedName{
+		Name:      p.opensearchCluster.Name,
+		Namespace: p.opensearchCluster.Namespace,
+	}, &loggingv1beta1.OpniOpensearch{}); err != nil {
+		if k8serrors.IsNotFound(err) {
+			p.logger.Info("opensearch cluster does not exist")
+			return nil, status.Error(codes.NotFound, "unable to list cluster status")
+		}
+		return nil, err
+	}
+
+	cluster := &opsterv1.OpenSearchCluster{}
+	if err := p.k8sClient.Get(ctx, types.NamespacedName{
+		Name:      p.opensearchCluster.Name,
+		Namespace: p.opensearchCluster.Namespace,
+	}, cluster); err != nil {
+		return nil, err
+	}
+
+	status := ClusterStatus(-1)
+
+	if !cluster.Status.Initialized {
+		status = ClusterStatusPending
+		return &loggingadmin.StatusResponse{
+			Status:  int32(status),
+			Details: ClusterStatusDescription(status),
+		}, nil
+	}
+
+	statusResp := p.opensearchManager.GetClusterStatus()
+	switch statusResp {
+	case opensearchdata.ClusterStatusGreen:
+		status = ClusterStatusGreen
+	case opensearchdata.ClusterStatusYellow:
+		status = ClusterStatusYellow
+	case opensearchdata.ClusterStatusRed:
+		status = ClusterStatusRed
+	case opensearchdata.ClusterStatusError:
+		status = ClusterStatusError
+	}
+
+	return &loggingadmin.StatusResponse{
+		Status:  int32(status),
+		Details: ClusterStatusDescription(status),
+	}, nil
+}
+
 func convertNodePoolToProtobuf(pool opsterv1.NodePool) (*loggingadmin.OpensearchNodeDetails, error) {
 	var tolerations []*corev1.Toleration
 	for _, toleration := range pool.Tolerations {
@@ -519,7 +596,7 @@ FETCH:
 		InsecureSkipVerify: true,
 	}
 
-	osCfg := opensearch.Config{
+	osCfg := osclient.Config{
 		Addresses: []string{
 			fmt.Sprintf("https://%s.%s:9200", cluster.Spec.General.ServiceName, cluster.Namespace),
 		},
@@ -529,7 +606,7 @@ FETCH:
 		Transport:            transport,
 	}
 
-	osClient, err := opensearch.NewClient(osCfg)
+	osClient, err := osclient.NewClient(osCfg)
 	if err != nil {
 		p.logger.Errorf("failed to create opensearch client: %v", err)
 		panic(err)

plugins/logging/pkg/gateway/plugin.go

@@ -39,7 +39,7 @@ import (
 	"github.com/rancher/opni/plugins/logging/pkg/apis/opensearch"
 	"github.com/rancher/opni/plugins/logging/pkg/backend"
 	"github.com/rancher/opni/plugins/logging/pkg/gateway/drivers"
-	loggingutil "github.com/rancher/opni/plugins/logging/pkg/util"
+	"github.com/rancher/opni/plugins/logging/pkg/opensearchdata"
 	corev1 "k8s.io/api/core/v1"
 )
 
@@ -60,7 +60,7 @@ type Plugin struct {
 	mgmtApi             future.Future[managementv1.ManagementClient]
 	nodeManagerClient   future.Future[capabilityv1.NodeManagerClient]
 	uninstallController future.Future[*task.Controller]
-	opensearchClient    *loggingutil.AsyncOpensearchClient
+	opensearchManager   opensearchdata.Manager
 	manageFlag          featureflags.FeatureFlag
 	logging             backend.LoggingBackend
 }
@@ -161,8 +161,10 @@ func NewPlugin(ctx context.Context, opts ...PluginOption) *Plugin {
 		storageBackend:      future.New[storage.Backend](),
 		mgmtApi:             future.New[managementv1.ManagementClient](),
 		uninstallController: future.New[*task.Controller](),
-		opensearchClient:    loggingutil.NewAsyncOpensearchClient(),
-		nodeManagerClient:   future.New[capabilityv1.NodeManagerClient](),
+		opensearchManager: *opensearchdata.NewManager(
+			lg.Named("opensearch-manager"),
+		),
+		nodeManagerClient: future.New[capabilityv1.NodeManagerClient](),
 	}
 
 	future.Wait4(p.storageBackend, p.mgmtApi, p.uninstallController, p.nodeManagerClient,
@@ -231,7 +233,7 @@ func Scheme(ctx context.Context) meta.Scheme {
 		p.manageFlag = p.featureOverride
 	}
 
-	go p.opensearchClient.SetClient(p.setOpensearchClient)
+	go p.opensearchManager.SetClient(p.setOpensearchClient)
 
 	scheme.Add(system.SystemPluginID, system.NewPlugin(p))
 	scheme.Add(httpext.HTTPAPIExtensionPluginID, httpext.NewPlugin(p))