fix: initialize chan operation

pkg/yurthub/cachemanager/cache_manager.go

@@ -97,17 +97,16 @@ func NewCacheManager(
 		listSelectorCollector: make(map[storage.Key]string),
 		inMemoryCache:         make(map[string]runtime.Object),
 		errorKeys: &errorKeys{
-			keys: make(map[string]string),
+			keys:       make(map[string]string),
+			operations: make(chan operation, 100),
 		},
 	}
 	cm.errorKeys.recover()
-
 	return cm
 }
 
 // CacheResponse cache response of request into backend storage
 func (cm *cacheManager) CacheResponse(req *http.Request, prc io.ReadCloser, stopCh <-chan struct{}) error {
-	go cm.errorKeys.sync(stopCh)
 	ctx := req.Context()
 	info, _ := apirequest.RequestInfoFrom(ctx)
 	if isWatch(ctx) {

pkg/yurthub/cachemanager/error_keys.go

@@ -18,6 +18,7 @@ package cachemanager
 
 import (
 	"bufio"
+	"context"
 	"encoding/json"
 	"os"
 	"path/filepath"
@@ -42,6 +43,7 @@ func NewErrorKeys() *errorKeys {
 		keys:       make(map[string]string),
 		operations: make(chan operation, 10),
 	}
+	go ek.sync(context.Background())
 	return ek
 }
 
@@ -89,7 +91,7 @@ func (ek *errorKeys) length() int {
 	return len(ek.keys)
 }
 
-func (ek *errorKeys) sync(stopChan <-chan struct{}) {
+func (ek *errorKeys) sync(ctx context.Context) {
 	err := os.MkdirAll(AOFPrefix, 0755)
 	if err != nil {
 		klog.Errorf("failed to create dir: %v", err)
@@ -108,7 +110,7 @@ func (ek *errorKeys) sync(stopChan <-chan struct{}) {
 			}
 			file.Write(append(data, '\n'))
 			file.Sync()
-		case <-stopChan:
+		case <-ctx.Done():
 			return
 		}
 	}

pkg/yurthub/cachemanager/error_keys_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package cachemanager
 
 import (
+	"context"
 	"errors"
 	"os"
 	"testing"
@@ -52,8 +53,8 @@ func TestXxx(t *testing.T) {
 				keys:       make(map[string]string),
 				operations: make(chan operation, 100),
 			}
-			stopChan := make(chan struct{})
-			go ek.sync(stopChan)
+			ctx, cancel := context.WithCancel(context.TODO())
+			go ek.sync(ctx)
 			for i := range tc.keys {
 				ek.put(tc.keys[i], tc.err[i])
 			}
@@ -69,7 +70,7 @@ func TestXxx(t *testing.T) {
 			if ek.length() != 0 {
 				t.Errorf("expect length %v, got %v", tc.length, ek.length())
 			}
-			close(stopChan)
+			cancel()
 			os.RemoveAll(AOFPrefix)
 		})
 	}