Merge tag 'v1.6.9' into release/1.11

# Conflicts:
#	actor/actor.go
#	actor/manager/container.go
#	actor/manager/manager.go
#	actor/mock/mock_server.go

actor/actor.go

@@ -42,7 +42,11 @@ type Server interface {
 	// SaveState is impl by ServerImplBase, It saves the state cache of this actor instance to state store component by calling api of daprd.
 	// Save state is called at two places: 1. On invocation of this actor instance. 2. When new actor starts.
 	SaveState() error
-
+	// Activate called when actor created by actor manager
+	Activate(invokeName string) error
+	// Deactivate called before actor removed by actor manager
+	Deactivate() error
+
 	WithContext() ServerContext
 }
 

actor/config/config.go

@@ -17,7 +17,11 @@ import "github.com/dapr/go-sdk/actor/codec/constant"
 
 // ActorConfig is Actor's configuration struct.
 type ActorConfig struct {
-	SerializerType string
+	SerializerType         string
+	ActorIdleTimeout       string
+	ActorScanInterval      string
+	DrainOngingCallTimeout string
+	DrainBalancedActors    bool
 }
 
 // Option is option function of ActorConfig.
@@ -30,6 +34,34 @@ func WithSerializerName(serializerType string) Option {
 	}
 }
 
+// WithActorIdleTimeout set actorIdleTimeout type of the actor as @actorIdleTimeout.
+func WithActorIdleTimeout(actorIdleTimeout string) Option {
+	return func(config *ActorConfig) {
+		config.ActorIdleTimeout = actorIdleTimeout
+	}
+}
+
+// WithActorScanInterval set actorScanInterval type of the actor as @actorScanInterval.
+func WithActorScanInterval(actorScanInterval string) Option {
+	return func(config *ActorConfig) {
+		config.ActorScanInterval = actorScanInterval
+	}
+}
+
+// WithDrainOngingCallTimeout set drainOngingCallTimeout type of the actor as @drainOngingCallTimeout.
+func WithDrainOngingCallTimeout(drainOngingCallTimeout string) Option {
+	return func(config *ActorConfig) {
+		config.DrainOngingCallTimeout = drainOngingCallTimeout
+	}
+}
+
+// WithDrainBalancedActors set drainBalancedActors type of the actor as @drainBalancedActors.
+func WithDrainBalancedActors(drainBalancedActors bool) Option {
+	return func(config *ActorConfig) {
+		config.DrainBalancedActors = drainBalancedActors
+	}
+}
+
 // GetConfigFromOptions get final ActorConfig set by @opts.
 func GetConfigFromOptions(opts ...Option) *ActorConfig {
 	conf := &ActorConfig{

actor/config/config_test.go

@@ -31,8 +31,16 @@ func TestRegisterActorTimer(t *testing.T) {
 	t.Run("get config with option", func(t *testing.T) {
 		config := GetConfigFromOptions(
 			WithSerializerName("mockSerializerType"),
+			WithActorIdleTimeout("1m"),
+			WithActorScanInterval("10s"),
+			WithDrainOngingCallTimeout("10s"),
+			WithDrainBalancedActors(true),
 		)
 		assert.NotNil(t, config)
 		assert.Equal(t, "mockSerializerType", config.SerializerType)
+		assert.Equal(t, "1m", config.ActorIdleTimeout)
+		assert.Equal(t, "10s", config.ActorScanInterval)
+		assert.Equal(t, "10s", config.DrainOngingCallTimeout)
+		assert.Equal(t, true, config.DrainBalancedActors)
 	})
 }

actor/manager/container.go

@@ -30,6 +30,7 @@ type ActorContainer interface {
 	Invoke(methodName string, param []byte) ([]reflect.Value, actorErr.ActorErr)
 	//nolint:staticcheck // SA1019 Deprecated: use ActorContainerContext instead.
 	GetActor() actor.Server
+	Deactivate() error
 }
 
 type ActorContainerContext interface {
@@ -79,12 +80,17 @@ func NewDefaultActorContainerContext(ctx context.Context, actorID string, impl a
 	impl.SetID(actorID)
 	daprClient, _ := dapr.NewClient()
 	// create state manager for this new actor
-	impl.SetStateManager(state.NewActorStateManagerContext(impl.Type(), actorID, state.NewDaprStateAsyncProvider(daprClient)))
-	// save state of this actor
-	err := impl.SaveState(ctx)
+	impl.SetStateManager(state.NewActorStateManager(impl.Type(), actorID, state.NewDaprStateAsyncProvider(daprClient)))
+	// move out for Activate param
+	/*err := impl.Activate(ctx)
 	if err != nil {
 		return nil, actorErr.ErrSaveStateFailed
 	}
+	// save state of this actor
+	err = impl.SaveState()
+	if err != nil {
+		return nil, actorErr.ErrSaveStateFailed
+	}*/
 	methodType, err := getAbsctractMethodMap(impl)
 	if err != nil {
 		log.Printf("failed to get absctract method map from registered provider, err = %s", err)
@@ -118,6 +124,10 @@ func (d *DefaultActorContainerContext) Invoke(ctx context.Context, methodName st
 	return returnValue, actorErr.Success
 }
 
+func (d *DefaultActorContainer) Deactivate() error {
+	return d.actor.Deactivate()
+}
+
 func (d *DefaultActorContainerContext) GetActor() actor.ServerContext {
 	return d.actor
 }

actor/manager/manager.go

@@ -52,6 +52,8 @@ type ActorManager interface {
 	DeactivateActor(actorID string) actorErr.ActorErr
 	InvokeReminder(actorID, reminderName string, params []byte) actorErr.ActorErr
 	InvokeTimer(actorID, timerName string, params []byte) actorErr.ActorErr
+	InvokeActors(methodName string, request []byte) actorErr.ActorErr
+	KillAllActors() actorErr.ActorErr
 }
 
 type ActorManagerContext interface {
@@ -127,13 +129,22 @@ func (m *DefaultActorManagerContext) RegisterActorImplFactory(f actor.FactoryCon
 }
 
 // getAndCreateActorContainerIfNotExist will.
-func (m *DefaultActorManagerContext) getAndCreateActorContainerIfNotExist(ctx context.Context, actorID string) (ActorContainerContext, actorErr.ActorErr) {
+func (m *DefaultActorManagerContext) getAndCreateActorContainerIfNotExist(ctx context.Context, actorID, invokeName string) (ActorContainerContext, actorErr.ActorErr) {
 	val, ok := m.activeActors.Load(actorID)
 	if !ok {
 		newContainer, aerr := NewDefaultActorContainerContext(ctx, actorID, m.factory(), m.serializer)
 		if aerr != actorErr.Success {
 			return nil, aerr
 		}
+		err := newContainer.GetActor().Activate(invokeName)
+		if err != nil {
+			return nil, actorErr.ErrSaveStateFailed
+		}
+		// save state of this actor
+		err = newContainer.GetActor().SaveState()
+		if err != nil {
+			return nil, actorErr.ErrSaveStateFailed
+		}
 		m.activeActors.Store(actorID, newContainer)
 		val, _ = m.activeActors.Load(actorID)
 	}
@@ -146,7 +157,7 @@ func (m *DefaultActorManagerContext) InvokeMethod(ctx context.Context, actorID,
 		return nil, actorErr.ErrActorFactoryNotSet
 	}
 
-	actorContainer, aerr := m.getAndCreateActorContainerIfNotExist(ctx, actorID)
+	actorContainer, aerr := m.getAndCreateActorContainerIfNotExist(ctx, actorID, methodName)
 	if aerr != actorErr.Success {
 		return nil, aerr
 	}
@@ -187,6 +198,7 @@ func (m *DefaultActorManagerContext) DeactivateActor(_ context.Context, actorID
 	if !ok {
 		return actorErr.ErrActorIDNotFound
 	}
+	actor.(ActorContainer).Deactivate()
 	m.activeActors.Delete(actorID)
 	return actorErr.Success
 }
@@ -201,7 +213,7 @@ func (m *DefaultActorManagerContext) InvokeReminder(ctx context.Context, actorID
 		log.Printf("failed to unmarshal reminder param, err: %v ", err)
 		return actorErr.ErrRemindersParamsInvalid
 	}
-	actorContainer, aerr := m.getAndCreateActorContainerIfNotExist(ctx, actorID)
+	actorContainer, aerr := m.getAndCreateActorContainerIfNotExist(ctx, actorID, reminderName)
 	if aerr != actorErr.Success {
 		return aerr
 	}
@@ -224,14 +236,55 @@ func (m *DefaultActorManagerContext) InvokeTimer(ctx context.Context, actorID, t
 		log.Printf("failed to unmarshal reminder param, err: %v ", err)
 		return actorErr.ErrTimerParamsInvalid
 	}
-	actorContainer, aerr := m.getAndCreateActorContainerIfNotExist(ctx, actorID)
+	actorContainer, aerr := m.getAndCreateActorContainerIfNotExist(ctx, actorID, timerName)
 	if aerr != actorErr.Success {
 		return aerr
 	}
 	_, aerr = actorContainer.Invoke(ctx, timerParams.CallBack, timerParams.Data)
 	return aerr
 }
 
+func (m *DefaultActorManager) InvokeActors(methodName string, request []byte) actorErr.ActorErr {
+	m.activeActors.Range(func(key, value interface{}) bool {
+		return func() bool {
+			go func() {
+				defer func() {
+					if err := recover(); err != nil {
+						log.Printf("InvokeActors recover, methodName:%s, request:%s", methodName, string(request))
+					}
+				}()
+				out, err := m.InvokeMethod(key.(string), methodName, request)
+				if err != actorErr.Success {
+					log.Printf("InvokeActors, methodName:%s, request:%s, out:%s, err:%v", methodName, string(request), string(out), err)
+				}
+			}()
+			return true
+		}()
+	})
+	return actorErr.Success
+}
+
+func (m *DefaultActorManager) KillAllActors() actorErr.ActorErr {
+	var actorIds []string
+	m.activeActors.Range(func(key, value interface{}) bool {
+		return func() bool {
+			actorIds = append(actorIds, key.(string))
+			return true
+		}()
+	})
+	for _, actorId := range actorIds {
+		func() {
+			defer func() {
+				if err := recover(); err != nil {
+					log.Printf("KillAllActors recover, actorId:%s", actorId)
+				}
+			}()
+			m.DeactivateActor(actorId)
+		}()
+	}
+	return actorErr.Success
+}
+
 func getAbsctractMethodMap(rcvr interface{}) (map[string]*MethodType, error) {
 	s := &Service{}
 	s.reflectType = reflect.TypeOf(rcvr)

actor/mock/mock_factory_impl.go

@@ -28,6 +28,14 @@ type ActorImpl struct {
 	actor.ServerImplBase
 }
 
+func (t *ActorImpl) Activate(invokeName string) error {
+	return nil
+}
+
+func (t *ActorImpl) Deactivate() error {
+	return nil
+}
+
 func (t *ActorImpl) Type() string {
 	return "testActorType"
 }
@@ -70,6 +78,14 @@ type NotReminderCalleeActor struct {
 	actor.ServerImplBaseCtx
 }
 
+func (t *NotReminderCalleeActor) Activate() error {
+	return nil
+}
+
+func (t *NotReminderCalleeActor) Deactivate() error {
+	return nil
+}
+
 func (t *NotReminderCalleeActor) Type() string {
 	return "testActorNotReminderCalleeType"
 }

actor/runtime/actor_runtime.go

@@ -16,6 +16,7 @@ package runtime
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"sync"
 
 	"github.com/dapr/go-sdk/actor"
@@ -73,6 +74,10 @@ func (r *ActorRunTimeContext) RegisterActorFactory(f actor.FactoryContext, opt .
 	conf := config.GetConfigFromOptions(opt...)
 	actType := f().Type()
 	r.config.RegisteredActorTypes = append(r.config.RegisteredActorTypes, actType)
+	r.config.ActorIdleTimeout = conf.ActorIdleTimeout
+	r.config.ActorScanInterval = conf.ActorScanInterval
+	r.config.DrainOngingCallTimeout = conf.DrainOngingCallTimeout
+	r.config.DrainBalancedActors = conf.DrainBalancedActors
 	mng, ok := r.actorManagers.Load(actType)
 	if !ok {
 		newMng, err := manager.NewDefaultActorManagerContext(conf.SerializerType)
@@ -154,3 +159,20 @@ func (r *ActorRunTime) InvokeReminder(actorTypeName, actorID, reminderName strin
 func (r *ActorRunTime) InvokeTimer(actorTypeName, actorID, timerName string, params []byte) actorErr.ActorErr {
 	return r.ctx.InvokeTimer(context.Background(), actorTypeName, actorID, timerName, params)
 }
+
+func (r *ActorRunTime) InvokeActors(actorType, methodName string, request []byte) actorErr.ActorErr {
+	mng, ok := r.actorManagers.Load(actorType)
+	if !ok {
+		return actorErr.ErrActorTypeNotFound
+	}
+	return mng.(manager.ActorManager).InvokeActors(methodName, request)
+}
+
+func (r *ActorRunTime) KillAllActors(actorTypeName string) actorErr.ActorErr {
+	fmt.Println("KillAllActors:", actorTypeName)
+	targetManager, ok := r.actorManagers.Load(actorTypeName)
+	if !ok {
+		return actorErr.ErrActorTypeNotFound
+	}
+	return targetManager.(manager.ActorManager).KillAllActors()
+}

client/configuration.go

@@ -9,6 +9,10 @@ import (
 	pb "github.com/dapr/dapr/pkg/proto/runtime/v1"
 )
 
+var (
+	ErrEmpty = errors.New("get configuration item result empty")
+)
+
 type ConfigurationItem struct {
 	Value    string
 	Version  string
@@ -29,7 +33,7 @@ func (c *GRPCClient) GetConfigurationItem(ctx context.Context, storeName, key st
 		return nil, err
 	}
 	if len(items) == 0 {
-		return nil, nil
+		return nil, ErrEmpty
 	}
 
 	return items[key], nil

service/common/type.go

@@ -15,6 +15,7 @@ package common
 
 import (
 	"encoding/json"
+	"net/http"
 )
 
 // TopicEvent is the content of the inbound topic message.
@@ -66,7 +67,8 @@ type InvocationEvent struct {
 	// Verb is the HTTP verb that was used to invoke this service.
 	Verb string `json:"-"`
 	// QueryString represents an encoded HTTP url query string in the following format: name=value&name2=value2
-	QueryString string `json:"-"`
+	QueryString string        `json:"-"`
+	Request     *http.Request `json:"-"`
 }
 
 // Content is a generic data content.

service/http/health_check.go

@@ -38,7 +38,8 @@ func (s *Server) AddHealthCheckHandler(route string, fn common.HealthCheckHandle
 				return
 			}
 
-			w.WriteHeader(http.StatusNoContent)
+			// fix check fail on k8s, http.StatusNoContent -> http.StatusOK
+			w.WriteHeader(http.StatusOK)
 		})))
 
 	return nil

service/http/invoke.go

@@ -52,6 +52,7 @@ func (s *Server) AddServiceInvocationHandler(route string, fn common.ServiceInvo
 				Verb:        r.Method,
 				QueryString: r.URL.RawQuery,
 				ContentType: r.Header.Get("Content-type"),
+				Request:     r,
 			}
 
 			var err error