feat: Support all component health check (#1094)

Co-authored-by: wenxuwan <wangwx_junction@163.com>
mosn · Dec 17, 2024 · 39c3fa4 · 39c3fa4
1 parent 20b531e
commit 39c3fa4
Show file tree

Hide file tree

Showing 29 changed files with 733 additions and 21 deletions.
diff --git a/components/configstores/etcdv3/etcdv3.go b/components/configstores/etcdv3/etcdv3.go
@@ -10,6 +10,7 @@
 // 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 etcdv3
 
 import (
@@ -22,6 +23,8 @@ import (
 
 	"mosn.io/pkg/utils"
 
+	"mosn.io/layotto/components/pkg/actuators"
+
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"mosn.io/pkg/log"
@@ -31,10 +34,22 @@ import (
 )
 
 const (
-	defaultGroup = "default"
-	defaultLabel = "default"
+	defaultGroup  = "default"
+	defaultLabel  = "default"
+	componentName = "configstore-etcdv3"
 )
 
+var (
+	once               sync.Once
+	readinessIndicator *actuators.HealthIndicator
+	livenessIndicator  *actuators.HealthIndicator
+)
+
+func init() {
+	readinessIndicator = actuators.NewHealthIndicator()
+	livenessIndicator = actuators.NewHealthIndicator()
+}
+
 type EtcdV3ConfigStore struct {
 	client *clientv3.Client
 	sync.RWMutex
@@ -56,6 +71,10 @@ func (c *EtcdV3ConfigStore) GetDefaultLabel() string {
 }
 
 func NewStore() configstores.Store {
+	once.Do(func() {
+		indicators := &actuators.ComponentsIndicator{ReadinessIndicator: readinessIndicator, LivenessIndicator: livenessIndicator}
+		actuators.SetComponentsIndicator(componentName, indicators)
+	})
 	return &EtcdV3ConfigStore{subscribeKey: make(map[string]string), watchRespCh: make(chan *configstores.SubscribeResp)}
 }
 
@@ -71,6 +90,12 @@ func (c *EtcdV3ConfigStore) Init(config *configstores.StoreConfig) error {
 		DialTimeout: time.Duration(t) * time.Second,
 	})
 	c.storeName = config.StoreName
+	if err != nil {
+		readinessIndicator.ReportError(err.Error())
+		livenessIndicator.ReportError(err.Error())
+	}
+	readinessIndicator.SetStarted()
+	livenessIndicator.SetStarted()
 	return err
 }
 

diff --git a/components/configstores/nacos/configstore.go b/components/configstores/nacos/configstore.go
@@ -30,8 +30,24 @@ import (
 	"mosn.io/pkg/log"
 
 	"mosn.io/layotto/components/configstores"
+	"mosn.io/layotto/components/pkg/actuators"
 )
 
+const (
+	componentName = "configstore-nacos"
+)
+
+var (
+	once               sync.Once
+	readinessIndicator *actuators.HealthIndicator
+	livenessIndicator  *actuators.HealthIndicator
+)
+
+func init() {
+	readinessIndicator = actuators.NewHealthIndicator()
+	livenessIndicator = actuators.NewHealthIndicator()
+}
+
 type ConfigStore struct {
 	client      config_client.IConfigClient
 	storeName   string
@@ -41,6 +57,10 @@ type ConfigStore struct {
 }
 
 func NewStore() configstores.Store {
+	once.Do(func() {
+		indicators := &actuators.ComponentsIndicator{ReadinessIndicator: readinessIndicator, LivenessIndicator: livenessIndicator}
+		actuators.SetComponentsIndicator(componentName, indicators)
+	})
 	return &ConfigStore{}
 }
 
@@ -91,10 +111,14 @@ func (n *ConfigStore) Init(config *configstores.StoreConfig) (err error) {
 	} else {
 		client, err = n.init(config.Address, timeoutMs, metadata)
 	}
+
 	if err != nil {
+		readinessIndicator.ReportError(err.Error())
+		livenessIndicator.ReportError(err.Error())
 		return err
 	}
-
+	readinessIndicator.SetStarted()
+	livenessIndicator.SetStarted()
 	n.client = client
 	// replace nacos sdk log
 	return n.setupLogger(metadata)

diff --git a/components/cryption/aliyun/kms.go b/components/cryption/aliyun/kms.go
diff --git a/components/cryption/aws/kms.go b/components/cryption/aws/kms.go
diff --git a/components/delay_queue/azure/servicebus/servicebus.go b/components/delay_queue/azure/servicebus/servicebus.go
@@ -10,6 +10,7 @@
 // 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 servicebus
 
 import (

diff --git a/components/file/aliyun/file.go b/components/file/aliyun/file.go
@@ -22,6 +22,9 @@ import (
 	"fmt"
 	"io"
 	"strconv"
+	"sync"
+
+	"mosn.io/layotto/components/pkg/actuators"
 
 	"github.com/aliyun/aliyun-oss-go-sdk/oss"
 
@@ -32,14 +35,30 @@ import (
 
 const (
 	storageTypeKey = "storageType"
+	componentName  = "file-aliyun"
+)
+
+var (
+	once               sync.Once
+	readinessIndicator *actuators.HealthIndicator
+	livenessIndicator  *actuators.HealthIndicator
 )
 
+func init() {
+	readinessIndicator = actuators.NewHealthIndicator()
+	livenessIndicator = actuators.NewHealthIndicator()
+}
+
 // AliyunFile is a binding for an AliCloud OSS storage bucketKey
 type AliyunFile struct {
 	client *oss.Client
 }
 
 func NewAliyunFile() file.File {
+	once.Do(func() {
+		indicators := &actuators.ComponentsIndicator{ReadinessIndicator: readinessIndicator, LivenessIndicator: livenessIndicator}
+		actuators.SetComponentsIndicator(componentName, indicators)
+	})
 	oss := &AliyunFile{}
 	return oss
 }
@@ -49,6 +68,8 @@ func (s *AliyunFile) Init(ctx context.Context, metadata *file.FileConfig) error
 	m := make([]*utils.OssMetadata, 0)
 	err := json.Unmarshal(metadata.Metadata, &m)
 	if err != nil {
+		readinessIndicator.ReportError(err.Error())
+		livenessIndicator.ReportError(err.Error())
 		return file.ErrInvalid
 	}
 
@@ -58,10 +79,14 @@ func (s *AliyunFile) Init(ctx context.Context, metadata *file.FileConfig) error
 		}
 		client, err := oss.New(v.Endpoint, v.AccessKeyID, v.AccessKeySecret)
 		if err != nil {
+			readinessIndicator.ReportError(err.Error())
+			livenessIndicator.ReportError(err.Error())
 			return err
 		}
 		s.client = client
 	}
+	readinessIndicator.SetStarted()
+	livenessIndicator.SetStarted()
 	return nil
 }
 

diff --git a/components/file/aws/file.go b/components/file/aws/file.go
@@ -23,7 +23,9 @@ import (
 	"fmt"
 	"io"
 	"strings"
+	"sync"
 
+	"mosn.io/layotto/components/pkg/actuators"
 	"mosn.io/layotto/components/pkg/utils"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -38,14 +40,30 @@ import (
 
 const (
 	defaultCredentialsSource = "provider"
+	componentName            = "file-aws"
 )
 
+var (
+	once               sync.Once
+	readinessIndicator *actuators.HealthIndicator
+	livenessIndicator  *actuators.HealthIndicator
+)
+
+func init() {
+	readinessIndicator = actuators.NewHealthIndicator()
+	livenessIndicator = actuators.NewHealthIndicator()
+}
+
 // AwsOss is a binding for aws oss storage.
 type AwsOss struct {
 	client *s3.Client
 }
 
 func NewAwsFile() file.File {
+	once.Do(func() {
+		indicators := &actuators.ComponentsIndicator{ReadinessIndicator: readinessIndicator, LivenessIndicator: livenessIndicator}
+		actuators.SetComponentsIndicator(componentName, indicators)
+	})
 	return &AwsOss{}
 }
 
@@ -54,6 +72,8 @@ func (a *AwsOss) Init(ctx context.Context, config *file.FileConfig) error {
 	m := make([]*utils.OssMetadata, 0)
 	err := json.Unmarshal(config.Metadata, &m)
 	if err != nil {
+		readinessIndicator.ReportError(err.Error())
+		livenessIndicator.ReportError(err.Error())
 		return errors.New("invalid config for aws oss")
 	}
 	for _, data := range m {
@@ -64,6 +84,10 @@ func (a *AwsOss) Init(ctx context.Context, config *file.FileConfig) error {
 		if err != nil {
 			continue
 		}
+		if client != nil {
+			readinessIndicator.SetStarted()
+			livenessIndicator.SetStarted()
+		}
 		a.client = client
 	}
 	return nil