fix pipeline stuck in queue status because etcd server abnormal

Author: chengjoey

internal/tools/pipeline/providers/dbgc/definition_cleanup/provider.go

@@ -69,7 +69,7 @@ func (p *provider) handleLogDir() error {
 		return err
 	}
 	// dir is not exist
-	return os.Mkdir(p.Cfg.LogDir, 0755)
+	return os.MkdirAll(p.Cfg.LogDir, 0755)
 }
 
 func (p *provider) Init(ctx servicehub.Context) error {

internal/tools/pipeline/providers/leaderworker/impl_worker.go

@@ -16,8 +16,10 @@ package leaderworker
 
 import (
 	"context"
+	"fmt"
 	"time"
 
+	"github.com/pkg/errors"
 	clientv3 "go.etcd.io/etcd/client/v3"
 
 	"github.com/erda-project/erda/internal/tools/pipeline/providers/leaderworker/lwctx"
@@ -151,6 +153,10 @@ func (p *provider) workerIntervalCleanupOnDelete(ctx context.Context, ev Event)
 			if err == nil {
 				return
 			}
+			// actively panic exit
+			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+				panic(fmt.Errorf("actively panic exit, maybe the etcd server has been shut down, err: %v", err))
+			}
 			p.Log.Errorf("failed to do worker interval cleanup on delete(auto retry), step: delete heartbeat key, workerID: %s, err: %v", ev.WorkerID, err)
 			time.Sleep(p.Cfg.Worker.RetryInterval)
 		}