Merge branch 'master' into fix_owner_block_http_api

cdc/capture/capture.go

@@ -28,6 +28,7 @@ import (
 	"github.com/pingcap/ticdc/cdc/model"
 	"github.com/pingcap/ticdc/cdc/owner"
 	"github.com/pingcap/ticdc/cdc/processor"
+	"github.com/pingcap/ticdc/cdc/processor/pipeline/system"
 	"github.com/pingcap/ticdc/pkg/config"
 	cdcContext "github.com/pingcap/ticdc/pkg/context"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
@@ -60,6 +61,8 @@ type Capture struct {
 	etcdClient *etcd.CDCEtcdClient
 	grpcPool   kv.GrpcPool
 
+	tableActorSystem *system.System
+
 	cancel context.CancelFunc
 
 	newProcessorManager func() *processor.Manager
@@ -103,6 +106,19 @@ func (c *Capture) reset(ctx context.Context) error {
 	if c.grpcPool != nil {
 		c.grpcPool.Close()
 	}
+	if c.tableActorSystem != nil {
+		err := c.tableActorSystem.Stop()
+		if err != nil {
+			log.Warn("stop table actor system failed", zap.Error(err))
+		}
+	}
+	if conf.Debug.EnableTableActor {
+		c.tableActorSystem = system.NewSystem()
+		err = c.tableActorSystem.Start(ctx)
+		if err != nil {
+			return errors.Annotate(cerror.WrapError(cerror.ErrNewCaptureFailed, err), "create capture session")
+		}
+	}
 	c.grpcPool = kv.NewGrpcPoolImpl(ctx, conf.Security)
 	log.Info("init capture", zap.String("capture-id", c.info.ID), zap.String("capture-addr", c.info.AdvertiseAddr))
 	return nil
@@ -148,11 +164,12 @@ func (c *Capture) Run(ctx context.Context) error {
 
 func (c *Capture) run(stdCtx context.Context) error {
 	ctx := cdcContext.NewContext(stdCtx, &cdcContext.GlobalVars{
-		PDClient:    c.pdClient,
-		KVStorage:   c.kvStorage,
-		CaptureInfo: c.info,
-		EtcdClient:  c.etcdClient,
-		GrpcPool:    c.grpcPool,
+		PDClient:         c.pdClient,
+		KVStorage:        c.kvStorage,
+		CaptureInfo:      c.info,
+		EtcdClient:       c.etcdClient,
+		GrpcPool:         c.grpcPool,
+		TableActorSystem: c.tableActorSystem,
 	})
 	err := c.register(ctx)
 	if err != nil {
@@ -354,6 +371,13 @@ func (c *Capture) AsyncClose() {
 	if c.grpcPool != nil {
 		c.grpcPool.Close()
 	}
+	if c.tableActorSystem != nil {
+		err := c.tableActorSystem.Stop()
+		if err != nil {
+			log.Warn("stop table actor system failed", zap.Error(err))
+		}
+		c.tableActorSystem = nil
+	}
 }
 
 // WriteDebugInfo writes the debug info into writer.

cdc/metrics.go

@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/ticdc/cdc/sink"
 	"github.com/pingcap/ticdc/cdc/sorter"
 	"github.com/pingcap/ticdc/cdc/sorter/leveldb"
+	"github.com/pingcap/ticdc/cdc/sorter/leveldb/system"
 	"github.com/pingcap/ticdc/cdc/sorter/memory"
 	"github.com/pingcap/ticdc/cdc/sorter/unified"
 	"github.com/pingcap/ticdc/pkg/actor"
@@ -53,4 +54,5 @@ func init() {
 	memory.InitMetrics(registry)
 	unified.InitMetrics(registry)
 	leveldb.InitMetrics(registry)
+	system.InitMetrics(registry)
 }

cdc/processor/pipeline/system/system.go

@@ -0,0 +1,44 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package system
+
+import (
+	"context"
+
+	"github.com/pingcap/ticdc/pkg/actor"
+)
+
+// System manages table pipeline global resource.
+type System struct {
+	tableActorSystem *actor.System
+	tableActorRouter *actor.Router
+}
+
+// NewSystem returns a system.
+func NewSystem() *System {
+	return &System{}
+}
+
+// Start starts a system.
+func (s *System) Start(ctx context.Context) error {
+	// todo: make the table actor system configurable
+	s.tableActorSystem, s.tableActorRouter = actor.NewSystemBuilder("table").Build()
+	s.tableActorSystem.Start(ctx)
+	return nil
+}
+
+// Stop stops a system.
+func (s *System) Stop() error {
+	return s.tableActorSystem.Stop()
+}

cdc/processor/pipeline/system/system_test.go

@@ -0,0 +1,29 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package system
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestStartAndStopSystem(t *testing.T) {
+	t.Parallel()
+
+	s := NewSystem()
+	require.Nil(t, s.Start(context.TODO()))
+	require.Nil(t, s.Stop())
+}