fix: no such container means WAL has been handled (#348)

Co-authored-by: anrs <anders.hu@shopee.com>

Author: anrs

cluster/calcium/calcium.go

@@ -1,6 +1,7 @@
 package calcium
 
 import (
+	"context"
 	"strings"
 
 	"github.com/projecteru2/core/cluster"
@@ -66,6 +67,11 @@ func New(config types.Config, embeddedStorage bool) (*Calcium, error) {
 	return cal, err
 }
 
+// DisasterRecover .
+func (c *Calcium) DisasterRecover(ctx context.Context) {
+	c.wal.Recover(ctx)
+}
+
 // Finalizer use for defer
 func (c *Calcium) Finalizer() {
 	c.store.TerminateEmbededStorage()

cluster/calcium/wal.go

@@ -3,6 +3,7 @@ package calcium
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"strings"
 
 	"github.com/projecteru2/core/log"
@@ -125,17 +126,24 @@ func (h *CreateWorkloadHandler) Handle(raw interface{}) error {
 	// There hasn't been the exact workload metadata, so we must remove it.
 	node, err := h.calcium.GetNode(context.Background(), wrk.Nodename)
 	if err != nil {
-		log.Errorf("[CreateWorkloadHandler.Check] Get node %s failed: %v", wrk.Nodename, err)
+		log.Errorf("[CreateWorkloadHandler.Handle] Get node %s failed: %v", wrk.Nodename, err)
 		return err
 	}
 	wrk.Engine = node.Engine
 
-	err = wrk.Remove(context.Background(), true)
-	if err != nil {
-		log.Errorf("[CreateWorkloadHandler.Check] Remove %s failed: %v", wrk.ID, err)
+	if err := wrk.Remove(context.Background(), true); err != nil {
+		if strings.HasPrefix(err.Error(), fmt.Sprintf("Error: No such container: %s", wrk.ID)) {
+			log.Errorf("[CreateWorkloadHandler.Handle] %s has been removed yet", wrk.ID)
+			return nil
+		}
+
+		log.Errorf("[CreateWorkloadHandler.Handle] Remove %s failed: %v", wrk.ID, err)
+		return err
 	}
 
-	return err
+	log.Warnf("[CreateWorkloadHandler.Handle] %s has been removed", wrk.ID)
+
+	return nil
 }
 
 // CreateLambdaHandler indicates event handler for creating lambda.

cluster/calcium/wal_test.go

@@ -73,11 +73,15 @@ func TestHandleCreateWorkloadError(t *testing.T) {
 	defer eng.AssertExpectations(t)
 	eng.On("VirtualizationRemove", mock.Anything, wrk.ID, true, true).
 		Return(fmt.Errorf("err")).
-		Twice()
+		Once()
+	c.wal.Recover(context.Background())
+
+	eng.On("VirtualizationRemove", mock.Anything, wrk.ID, true, true).
+		Return(fmt.Errorf("Error: No such container: %s", wrk.ID)).
+		Once()
 	c.wal.Recover(context.Background())
 
-	// Recovers again because the above recovering had been failed.
-	// so the log item hadn't been deleted.
+	// Nothing recovered.
 	c.wal.Recover(context.Background())
 }
 

core.go

@@ -75,6 +75,7 @@ func serve(c *cli.Context) error {
 		return err
 	}
 	defer cluster.Finalizer()
+	cluster.DisasterRecover(c.Context)
 
 	rpcch := make(chan struct{}, 1)
 	vibranium := rpc.New(cluster, config, rpcch)