feat: don't compile the actuator module if users don't need it

cmd/layotto/main.go

@@ -18,7 +18,6 @@ package main
 
 import (
 	"encoding/json"
-	"fmt"
 	"github.com/dapr/components-contrib/secretstores"
 	"github.com/dapr/components-contrib/secretstores/aws/parameterstore"
 	"github.com/dapr/components-contrib/secretstores/aws/secretmanager"
@@ -29,6 +28,7 @@ import (
 	secretstore_env "github.com/dapr/components-contrib/secretstores/local/env"
 	secretstore_file "github.com/dapr/components-contrib/secretstores/local/file"
 	"mosn.io/api"
+	component_actuators "mosn.io/layotto/components/pkg/actuators"
 	"mosn.io/layotto/diagnostics"
 	"mosn.io/layotto/pkg/grpc/default_api"
 	secretstores_loader "mosn.io/layotto/pkg/runtime/secretstores"
@@ -129,6 +129,7 @@ import (
 	"github.com/urfave/cli"
 	"google.golang.org/grpc"
 	_ "mosn.io/layotto/pkg/filter/network/tcpcopy"
+	l8_grpc "mosn.io/layotto/pkg/grpc"
 	"mosn.io/layotto/pkg/runtime"
 	_ "mosn.io/layotto/pkg/wasm"
 	"mosn.io/mosn/pkg/featuregate"
@@ -167,17 +168,35 @@ func init() {
 }
 
 func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrpc.RegisteredServer, error) {
+	var err error
+	defer func() {
+		if err != nil {
+			// fail fast if error occurs during startup.
+			// The reason we panic in a new goroutine is to prevent mosn from recovering.
+			go func() {
+				log.DefaultLogger.Errorf("An error occurred during startup : %v", err)
+				panic(err)
+			}()
+		}
+	}()
 	// 1. parse config
 	cfg, err := runtime.ParseRuntimeConfig(data)
 	if err != nil {
-		actuator.GetRuntimeReadinessIndicator().SetUnhealthy(fmt.Sprintf("parse config error.%v", err))
 		return nil, err
 	}
 	// 2. new instance
 	rt := runtime.NewMosnRuntime(cfg)
 	// 3. run
 	server, err := rt.Run(
 		runtime.WithGrpcOptions(opts...),
+		// wrap the grpc server with actuator
+		runtime.WithNewServer(func(apis []l8_grpc.GrpcAPI, opts ...grpc.ServerOption) (mgrpc.RegisteredServer, error) {
+			server, err := l8_grpc.NewDefaultServer(apis, opts...)
+			if err != nil {
+				return nil, err
+			}
+			return actuator.NewGrpcServerWithActuator(server)
+		}),
 		// register your grpc API here
 		runtime.WithGrpcAPI(
 			default_api.NewGrpcAPI,
@@ -376,11 +395,6 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 				return secretstore_env.NewEnvSecretStore(loggerForDaprComp)
 			}),
 		))
-	// 4. check if unhealthy
-	if err != nil {
-		actuator.GetRuntimeReadinessIndicator().SetUnhealthy(err.Error())
-		actuator.GetRuntimeLivenessIndicator().SetUnhealthy(err.Error())
-	}
 	return server, err
 }
 
@@ -417,16 +431,31 @@ var cmdStart = cli.Command{
 
 		stm.AppendStartStage(mosn.DefaultStartStage)
 
+		stm.AppendAfterStartStage(SetActuatorAfterStart)
+
 		stm.Run()
 
-		actuator.GetRuntimeReadinessIndicator().SetStarted()
-		actuator.GetRuntimeLivenessIndicator().SetStarted()
 		// wait mosn finished
 		stm.WaitFinish()
 		return nil
 	},
 }
 
+func SetActuatorAfterStart(m *mosn.Mosn) {
+	// register component actuator
+	component_actuators.RangeAllIndicators(
+		func(name string, v *component_actuators.ComponentsIndicator) bool {
+			if v != nil {
+				health.AddLivenessIndicator(name, v.LivenessIndicator)
+				health.AddReadinessIndicator(name, v.ReadinessIndicator)
+			}
+			return true
+		})
+	// set started
+	actuator.GetRuntimeReadinessIndicator().SetStarted()
+	actuator.GetRuntimeLivenessIndicator().SetStarted()
+}
+
 // ExtensionsRegister for register mosn rpc extensions
 func ExtensionsRegister(c *cli.Context) {
 	// 1. tracer driver register

cmd/layotto_multiple_api/main.go

@@ -18,9 +18,10 @@ package main
 
 import (
 	"encoding/json"
-	"fmt"
 	"mosn.io/api"
 	helloworld_api "mosn.io/layotto/cmd/layotto_multiple_api/helloworld"
+	component_actuators "mosn.io/layotto/components/pkg/actuators"
+	l8_grpc "mosn.io/layotto/pkg/grpc"
 	"mosn.io/layotto/pkg/grpc/dapr"
 	"mosn.io/layotto/pkg/grpc/default_api"
 	"os"
@@ -157,17 +158,35 @@ func init() {
 }
 
 func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrpc.RegisteredServer, error) {
+	var err error
+	defer func() {
+		if err != nil {
+			// fail fast if error occurs during startup.
+			// The reason we panic in a new goroutine is to prevent mosn from recovering.
+			go func() {
+				log.DefaultLogger.Errorf("An error occurred during startup : %v", err)
+				panic(err)
+			}()
+		}
+	}()
 	// 1. parse config
 	cfg, err := runtime.ParseRuntimeConfig(data)
 	if err != nil {
-		actuator.GetRuntimeReadinessIndicator().SetUnhealthy(fmt.Sprintf("parse config error.%v", err))
 		return nil, err
 	}
 	// 2. new instance
 	rt := runtime.NewMosnRuntime(cfg)
 	// 3. run
 	server, err := rt.Run(
 		runtime.WithGrpcOptions(opts...),
+		// wrap the grpc server with actuator
+		runtime.WithNewServer(func(apis []l8_grpc.GrpcAPI, opts ...grpc.ServerOption) (mgrpc.RegisteredServer, error) {
+			server, err := l8_grpc.NewDefaultServer(apis, opts...)
+			if err != nil {
+				return nil, err
+			}
+			return actuator.NewGrpcServerWithActuator(server)
+		}),
 		// register your grpc API here
 		runtime.WithGrpcAPI(
 			// default grpc API
@@ -340,11 +359,6 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 				return sequencer_zookeeper.NewZookeeperSequencer(log.DefaultLogger)
 			}),
 		))
-	// 4. check if unhealthy
-	if err != nil {
-		actuator.GetRuntimeReadinessIndicator().SetUnhealthy(err.Error())
-		actuator.GetRuntimeLivenessIndicator().SetUnhealthy(err.Error())
-	}
 	return server, err
 }
 
@@ -380,6 +394,8 @@ var cmdStart = cli.Command{
 
 		stm.AppendStartStage(mosn.DefaultStartStage)
 
+		stm.AppendAfterStartStage(SetActuatorAfterStart)
+
 		stm.Run()
 
 		actuator.GetRuntimeReadinessIndicator().SetStarted()
@@ -390,6 +406,21 @@ var cmdStart = cli.Command{
 	},
 }
 
+func SetActuatorAfterStart(m *mosn.Mosn) {
+	// register component actuator
+	component_actuators.RangeAllIndicators(
+		func(name string, v *component_actuators.ComponentsIndicator) bool {
+			if v != nil {
+				health.AddLivenessIndicator(name, v.LivenessIndicator)
+				health.AddReadinessIndicator(name, v.ReadinessIndicator)
+			}
+			return true
+		})
+	// set started
+	actuator.GetRuntimeReadinessIndicator().SetStarted()
+	actuator.GetRuntimeLivenessIndicator().SetStarted()
+}
+
 // ExtensionsRegister for register mosn rpc extensions
 func ExtensionsRegister(c *cli.Context) {
 	// 1. tracer driver register

components/configstores/apollo/indicator.go

@@ -35,7 +35,7 @@ func init() {
 	readinessIndicator = newHealthIndicator()
 	livenessIndicator = newHealthIndicator()
 	indicators := &actuators.ComponentsIndicator{ReadinessIndicator: readinessIndicator, LivenessIndicator: livenessIndicator}
-	actuators.SetComponentsActuators("apollo", indicators)
+	actuators.SetComponentsIndicator("apollo", indicators)
 }
 
 func newHealthIndicator() *healthIndicator {

components/go.sum

@@ -505,6 +505,7 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.10.4/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/klauspost/compress v1.12.2/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
+github.com/klauspost/compress v1.13.1/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
 github.com/klauspost/compress v1.13.5/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/klauspost/compress v1.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=

components/pkg/actuators/actutors.go

@@ -50,6 +50,12 @@ func GetIndicatorWithName(name string) *ComponentsIndicator {
 	return nil
 }
 
-func SetComponentsActuators(name string, indicator *ComponentsIndicator) {
+func SetComponentsIndicator(name string, indicator *ComponentsIndicator) {
 	componentsActutors.Store(name, indicator)
 }
+
+func RangeAllIndicators(f func(key string, value *ComponentsIndicator) bool) {
+	componentsActutors.Range(func(k, v interface{}) bool {
+		return f(k.(string), v.(*ComponentsIndicator))
+	})
+}

pkg/integrate/actuator/grpc.go

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2021 Layotto Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * 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 actuator
+
+import (
+	mgrpc "mosn.io/mosn/pkg/filter/network/grpc"
+	"net"
+)
+
+type GrpcServerWithActuator struct {
+	srv mgrpc.RegisteredServer
+}
+
+func (s *GrpcServerWithActuator) Serve(ln net.Listener) (err error) {
+	err = s.srv.Serve(ln)
+	if err != nil {
+		GetRuntimeReadinessIndicator().SetUnhealthy(err.Error())
+		GetRuntimeLivenessIndicator().SetUnhealthy(err.Error())
+	}
+	return
+}
+func (s *GrpcServerWithActuator) Stop() {
+	GetRuntimeReadinessIndicator().SetUnhealthy("shutdown")
+	GetRuntimeLivenessIndicator().SetUnhealthy("shutdown")
+	s.srv.Stop()
+}
+
+func (s *GrpcServerWithActuator) GracefulStop() {
+	GetRuntimeReadinessIndicator().SetUnhealthy("shutdown")
+	GetRuntimeLivenessIndicator().SetUnhealthy("shutdown")
+	s.srv.GracefulStop()
+}
+
+func NewGrpcServerWithActuator(srv mgrpc.RegisteredServer) (mgrpc.RegisteredServer, error) {
+	return &GrpcServerWithActuator{
+		srv: srv,
+	}, nil
+}

pkg/runtime/runtime.go

@@ -35,13 +35,10 @@ import (
 	"mosn.io/layotto/components/configstores"
 	"mosn.io/layotto/components/hello"
 	"mosn.io/layotto/components/lock"
-	"mosn.io/layotto/components/pkg/actuators"
 	"mosn.io/layotto/components/pkg/info"
 	"mosn.io/layotto/components/rpc"
 	"mosn.io/layotto/components/sequencer"
-	"mosn.io/layotto/pkg/actuator/health"
 	"mosn.io/layotto/pkg/grpc"
-	"mosn.io/layotto/pkg/integrate/actuator"
 	runtime_lock "mosn.io/layotto/pkg/runtime/lock"
 	runtime_pubsub "mosn.io/layotto/pkg/runtime/pubsub"
 	runtime_sequencer "mosn.io/layotto/pkg/runtime/sequencer"
@@ -200,8 +197,6 @@ func (m *MosnRuntime) Stop() {
 	if m.srv != nil {
 		m.srv.Stop()
 	}
-	actuator.GetRuntimeReadinessIndicator().SetUnhealthy("shutdown")
-	actuator.GetRuntimeLivenessIndicator().SetUnhealthy("shutdown")
 }
 
 func (m *MosnRuntime) initRuntime(o *runtimeOptions) error {
@@ -283,12 +278,6 @@ func (m *MosnRuntime) initConfigStores(configStores ...*configstores.StoreFactor
 			return err
 		}
 		m.configStores[name] = c
-		v := actuators.GetIndicatorWithName(name)
-		//Now don't force user implement actuator of components
-		if v != nil {
-			health.AddLivenessIndicator(name, v.LivenessIndicator)
-			health.AddReadinessIndicator(name, v.ReadinessIndicator)
-		}
 	}
 	return nil
 }
@@ -383,12 +372,6 @@ func (m *MosnRuntime) initFiles(files ...*file.FileFactory) error {
 			return err
 		}
 		m.files[name] = c
-		v := actuators.GetIndicatorWithName(name)
-		//Now don't force user implement actuator of components
-		if v != nil {
-			health.AddLivenessIndicator(name, v.LivenessIndicator)
-			health.AddReadinessIndicator(name, v.ReadinessIndicator)
-		}
 	}
 	return nil
 }