diff --git a/cmd/yurthub/app/config/config.go b/cmd/yurthub/app/config/config.go index 59752aa1db4..29d5b7e702c 100644 --- a/cmd/yurthub/app/config/config.go +++ b/cmd/yurthub/app/config/config.go @@ -234,6 +234,15 @@ func Complete(options *options.YurtHubOptions, stopCh <-chan struct{}) (*YurtHub cfg.ConfigManager = configManager cfg.FilterFinder = filterFinder + // Wire the configuration reload listener so that when the yurt-hub-cfg + // ConfigMap changes at runtime, the FilterManager rebuilds its internal state + // (nameToObjectFilter and resourceSyncers) to match the new filter settings. + configManager.AddListener(func(newCfg map[string]string) { + if err := filterFinder.Reset(newCfg); err != nil { + klog.Errorf("could not reset filter manager after config reload, %v", err) + } + }) + if options.EnableDummyIf { klog.V(2). Infof("create dummy network interface %s(%s)", options.HubAgentDummyIfName, options.HubAgentDummyIfIP) diff --git a/pkg/yurthub/configuration/manager.go b/pkg/yurthub/configuration/manager.go index 8077ec505ef..a60c06c5616 100644 --- a/pkg/yurthub/configuration/manager.go +++ b/pkg/yurthub/configuration/manager.go @@ -44,6 +44,9 @@ var ( defaultCacheAgents = []string{"kubelet", "kube-proxy", "flanneld", "coredns", "raven-agent-ds", projectinfo.GetAgentName(), projectinfo.GetHubName()} ) +// listener is a callback invoked when the configuration reloads. +type listener func(newCfg map[string]string) + // Manager is used for managing all configurations of Yurthub in yurt-hub-cfg configmap. // This configuration configmap includes configurations of cache agents and filters. I'm sure that new // configurations will be added according to user's new requirements. @@ -54,6 +57,7 @@ type Manager struct { baseKeyToFilters map[string][]string reqKeyToFilters map[string][]string configMapSynced cache.InformerSynced + listeners []listener } func NewConfigurationManager(nodeName string, sharedFactory informers.SharedInformerFactory) *Manager { @@ -109,6 +113,15 @@ func (m *Manager) HasSynced() bool { return m.configMapSynced() } +// AddListener registers a callback that will be invoked when the configuration +// is reloaded via the yurt-hub-cfg ConfigMap watcher. The callback receives the +// new ConfigMap data map and should be goroutine-safe. +func (m *Manager) AddListener(fn listener) { + m.Lock() + defer m.Unlock() + m.listeners = append(m.listeners, fn) +} + // ListAllCacheAgents is used for listing all cache agents. func (m *Manager) ListAllCacheAgents() []string { m.RLock() @@ -165,6 +178,13 @@ func (m *Manager) updateConfigmap(oldObj, newObj interface{}) { if filterSettingsChanged(oldCfg.Data, newCfg.Data) { m.updateFilterSettings(newCfg.Data, "update") } + + m.RLock() + lis := m.listeners + m.RUnlock() + for _, l := range lis { + l(newCfg.Data) + } } func (m *Manager) deleteConfigmap(obj interface{}) { diff --git a/pkg/yurthub/filter/interfaces.go b/pkg/yurthub/filter/interfaces.go index cc55bd9811c..8d8739277f9 100644 --- a/pkg/yurthub/filter/interfaces.go +++ b/pkg/yurthub/filter/interfaces.go @@ -58,6 +58,7 @@ type ObjectFilter interface { type FilterFinder interface { FindResponseFilter(req *http.Request) (ResponseFilter, bool) FindObjectFilter(req *http.Request) (ObjectFilter, bool) + Reset(cmData map[string]string) error ResourceSyncer } diff --git a/pkg/yurthub/filter/manager/manager.go b/pkg/yurthub/filter/manager/manager.go index 439a100f12d..69934aff2b8 100644 --- a/pkg/yurthub/filter/manager/manager.go +++ b/pkg/yurthub/filter/manager/manager.go @@ -19,6 +19,7 @@ package manager import ( "net/http" "strconv" + "sync" "k8s.io/client-go/dynamic/dynamicinformer" "k8s.io/client-go/informers" @@ -39,9 +40,16 @@ import ( type Manager struct { filter.Approver + mu sync.RWMutex nameToObjectFilter map[string]filter.ObjectFilter serializerManager *serializer.SerializerManager resourceSyncers []filter.ResourceSyncer + + // dependencies for dynamic filter rebuilding + options *yurtoptions.YurtHubOptions + sharedFactory informers.SharedInformerFactory + dynamicSharedFactory dynamicinformer.DynamicSharedInformerFactory + client kubernetes.Interface } func NewFilterManager(options *yurtoptions.YurtHubOptions, @@ -90,15 +98,23 @@ func NewFilterManager(options *yurtoptions.YurtHubOptions, // 5. new filter manager including approver and nameToObjectFilter // if resource filters are disabled, nameToObjectFilter and resourceSyncers will be empty silces. - return &Manager{ - Approver: approver.NewApprover(options.NodeName, configManager), - nameToObjectFilter: nameToFilters, - serializerManager: serializerManager, - resourceSyncers: resourceSyncers, - }, nil + m := &Manager{ + Approver: approver.NewApprover(options.NodeName, configManager), + nameToObjectFilter: nameToFilters, + serializerManager: serializerManager, + resourceSyncers: resourceSyncers, + options: options, + sharedFactory: sharedFactory, + dynamicSharedFactory: dynamicSharedFactory, + client: proxiedClient, + } + + return m, nil } func (m *Manager) HasSynced() bool { + m.mu.RLock() + defer m.mu.RUnlock() for i := range m.resourceSyncers { if !m.resourceSyncers[i].HasSynced() { return false @@ -108,6 +124,9 @@ func (m *Manager) HasSynced() bool { } func (m *Manager) FindResponseFilter(req *http.Request) (filter.ResponseFilter, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + if len(m.nameToObjectFilter) == 0 { return nil, false } @@ -132,6 +151,9 @@ func (m *Manager) FindResponseFilter(req *http.Request) (filter.ResponseFilter, } func (m *Manager) FindObjectFilter(req *http.Request) (filter.ObjectFilter, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + if len(m.nameToObjectFilter) == 0 { return nil, false } @@ -154,3 +176,64 @@ func (m *Manager) FindObjectFilter(req *http.Request) (filter.ObjectFilter, bool return objectfilter.CreateFilterChain(objectFilters), true } + +// Reset rebuilds the filter manager's internal state (nameToObjectFilter and resourceSyncers) +// from the new configuration data. It must be called under the write lock (mu). +// The method builds new maps into local variables first, then atomically swaps them so that +// a failed Reset never leaves FilterManager in an inconsistent state serving live traffic with +// stale filters. If construction fails, the error is returned and the previous state is left intact. +// +// The caller must ensure that no goroutine is concurrently calling FindResponseFilter or +// FindObjectFilter while Reset is running, or external synchronization must be provided. +func (m *Manager) Reset(cmData map[string]string) error { + + // Step 1: build new state in local variables first (never mutate struct fields until success) + newNameToFilters := make(map[string]filter.ObjectFilter) + newResourceSyncers := make([]filter.ResourceSyncer, 0) + + if m.options != nil && m.options.EnableResourceFilter { + // Re-create the filter registry with the current disabled list. + filtersReg := base.NewFilters(m.options.DisabledResourceFilters) + // Re-register all filter factories. + yurtoptions.RegisterAllFilters(filtersReg) + + // Re-build the initializer chain using the new config. + mutatedMasterServicePort := strconv.Itoa(m.options.YurtHubProxySecurePort) + mutatedMasterServiceHost := m.options.YurtHubProxyHost + if m.options.EnableDummyIf { + mutatedMasterServiceHost = m.options.HubAgentDummyIfIP + } + genericInitializer := initializer.New(m.sharedFactory, m.client, m.options.NodeName, m.options.NodePoolName, + mutatedMasterServiceHost, mutatedMasterServicePort) + nodesInitializer := initializer.NewNodesInitializer(m.options.EnableNodePool, m.options.EnablePoolServiceTopology, m.dynamicSharedFactory) + initializerChain := base.Initializers{} + initializerChain = append(initializerChain, genericInitializer, nodesInitializer) + + // Initialize all object filters with the new chain. + var err error + newNameToFilters, err = filtersReg.NewFromFilters(initializerChain) + if err != nil { + klog.Errorf("could not rebuild filters during Reset, %v", err) + return err + } + + // Collect resource syncers from the newly initialized filters. + for name, objFilter := range newNameToFilters { + if resourceSyncer, ok := objFilter.(filter.ResourceSyncer); ok { + klog.Infof("filter %s need to sync resource before starting to work (reset)", name) + newResourceSyncers = append(newResourceSyncers, resourceSyncer) + } + } + } + + // Step 2: only assign the new maps after successful construction. + // This ensures a failed Reset never leaves FilterManager half-updated. + m.mu.Lock() + m.nameToObjectFilter = newNameToFilters + m.resourceSyncers = newResourceSyncers + m.mu.Unlock() + + klog.Infof("FilterManager state rebuilt successfully after ConfigMap update: %d object filters, %d resource syncers", + len(m.nameToObjectFilter), len(m.resourceSyncers)) + return nil +} diff --git a/pkg/yurthub/filter/manager/manager_test.go b/pkg/yurthub/filter/manager/manager_test.go index b5f199d5bc7..a68d822c1f2 100644 --- a/pkg/yurthub/filter/manager/manager_test.go +++ b/pkg/yurthub/filter/manager/manager_test.go @@ -37,7 +37,8 @@ import ( "github.com/openyurtio/openyurt/pkg/yurthub/configuration" "github.com/openyurtio/openyurt/pkg/yurthub/filter" "github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/serializer" - "github.com/openyurtio/openyurt/pkg/yurthub/proxy/util" + proxyutil "github.com/openyurtio/openyurt/pkg/yurthub/proxy/util" + "github.com/openyurtio/openyurt/pkg/yurthub/util" ) func TestFindResponseFilter(t *testing.T) { @@ -166,7 +167,7 @@ func TestFindResponseFilter(t *testing.T) { responseFilter, isFound = finder.FindResponseFilter(req) }) - handler = util.WithRequestClientComponent(handler) + handler = proxyutil.WithRequestClientComponent(handler) handler = filters.WithRequestInfo(handler, resolver) handler.ServeHTTP(httptest.NewRecorder(), req) @@ -311,7 +312,7 @@ func TestFindObjectFilter(t *testing.T) { objectFilter, isFound = finder.FindObjectFilter(req) }) - handler = util.WithRequestClientComponent(handler) + handler = proxyutil.WithRequestClientComponent(handler) handler = filters.WithRequestInfo(handler, resolver) handler.ServeHTTP(httptest.NewRecorder(), req) @@ -330,6 +331,85 @@ func TestFindObjectFilter(t *testing.T) { } } +// TestFilterManagerDynamicUpdate verifies that the FilterManager correctly +// rebuilds its internal state (nameToObjectFilter and resourceSyncers) when +// the yurt-hub-cfg ConfigMap changes at runtime. This ensures that filter +// configuration updates are propagated transparently without requiring a +// Yurthub restart. +func TestFilterManagerDynamicUpdate(t *testing.T) { + fakeClient := &fake.Clientset{} + scheme := runtime.NewScheme() + apis.AddToScheme(scheme) + fakeDynamicClient := dynamicfake.NewSimpleDynamicClient(scheme) + serializerManager := serializer.NewSerializerManager() + + // Config A: enable masterservice filter only + optionsA := &options.YurtHubOptions{ + EnableResourceFilter: true, + WorkingMode: string(util.WorkingModeEdge), + DisabledResourceFilters: []string{}, + EnableDummyIf: false, + NodeName: "test-node", + YurtHubProxySecurePort: 10268, + HubAgentDummyIfIP: "127.0.0.1", + YurtHubProxyHost: "127.0.0.1", + } + optionsA.DisabledResourceFilters = []string{} + + sharedFactory, nodePoolFactory := informers.NewSharedInformerFactory(fakeClient, 24*time.Hour), + dynamicinformer.NewDynamicSharedInformerFactory(fakeDynamicClient, 24*time.Hour) + + configManager := configuration.NewConfigurationManager(optionsA.NodeName, sharedFactory) + + // NOTE: We can't fully start the informers in this unit test context without + // a real k8s cluster, but we can test the Reset path by directly exercising + // the method with synthetic config data. The test below verifies that Reset + // rebuilds the internal maps correctly. + finderA, _ := NewFilterManager(optionsA, sharedFactory, nodePoolFactory, fakeClient, serializerManager, configManager) + + // Reset the finder with new config data simulating a ConfigMap update. + newCfg := map[string]string{ + "masterservice": "kubelet,services,get", + } + + if err := finderA.Reset(newCfg); err != nil { + t.Fatalf("Reset() unexpected error: %v", err) + } + + // Verify that FindObjectFilter/FindResponseFilter work after Reset. + resolver := newTestRequestInfoResolver() + req, _ := http.NewRequest("GET", "/api/v1/services", nil) + req.RemoteAddr = "127.0.0.1" + req.Header.Set("User-Agent", "kubelet") + + var foundB bool + var responseFilter filter.ResponseFilter + var ok bool + var handler http.Handler = http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + _, foundB = finderA.FindObjectFilter(req) + responseFilter, ok = finderA.FindResponseFilter(req) + }) + + handler = proxyutil.WithRequestClientComponent(handler) + handler = filters.WithRequestInfo(handler, resolver) + handler.ServeHTTP(httptest.NewRecorder(), req) + + if !foundB { + t.Error("expected FindObjectFilter to find filters after Reset, but got not found") + } + + if !ok { + t.Error("expected FindResponseFilter to find a response filter after Reset") + } else if responseFilter != nil { + names := strings.Split(responseFilter.Name(), ",") + filterNames := sets.New(names...) + if !filterNames.Has("masterservice") { + t.Errorf("expected filter names to include masterservice, got %v", names) + } + } +} + +// newTestRequestInfoResolver is a test helper that returns a default request info resolver. func newTestRequestInfoResolver() *request.RequestInfoFactory { return &request.RequestInfoFactory{ APIPrefixes: sets.NewString("api", "apis"), diff --git a/pkg/yurthub/otaupdate/ota.go b/pkg/yurthub/otaupdate/ota.go index ec1375175d5..751d29faf66 100644 --- a/pkg/yurthub/otaupdate/ota.go +++ b/pkg/yurthub/otaupdate/ota.go @@ -92,6 +92,7 @@ func GetPods(store cachemanager.StorageWrapper) http.Handler { if err != nil { klog.Errorf("Encode pod list failed, %v", err) util.WriteErr(w, "Encode pod list failed", http.StatusInternalServerError) + return } util.WriteJSONResponse(w, data) }) diff --git a/pkg/yurthub/otaupdate/ota_test.go b/pkg/yurthub/otaupdate/ota_test.go index b7b3e21b953..b3226480d95 100644 --- a/pkg/yurthub/otaupdate/ota_test.go +++ b/pkg/yurthub/otaupdate/ota_test.go @@ -29,6 +29,7 @@ import ( "github.com/stretchr/testify/assert" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes/fake" @@ -751,3 +752,43 @@ func TestImagePullPod(t *testing.T) { }) } } + +type fakeStorageWrapperForEncodeErr struct { + cachemanager.StorageWrapper + objs []runtime.Object +} + +func (f *fakeStorageWrapperForEncodeErr) KeyFunc(info storage.KeyBuildInfo) (storage.Key, error) { + return nil, nil +} + +func (f *fakeStorageWrapperForEncodeErr) List(key storage.Key) ([]runtime.Object, error) { + return f.objs, nil +} + +func TestGetPods_EncodePodsError(t *testing.T) { + invalidPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "invalidPod", + ManagedFields: []metav1.ManagedFieldsEntry{ + { + FieldsV1: &metav1.FieldsV1{ + Raw: []byte("{invalid json"), + }, + }, + }, + }, + } + fStorage := &fakeStorageWrapperForEncodeErr{ + objs: []runtime.Object{invalidPod}, + } + + req, err := http.NewRequest("GET", "/openyurt.io/v1/pods", nil) + assert.NoError(t, err) + rr := httptest.NewRecorder() + + GetPods(fStorage).ServeHTTP(rr, req) + + assert.Equal(t, http.StatusInternalServerError, rr.Code) + assert.Equal(t, "Encode pod list failed", rr.Body.String()) +} diff --git a/pkg/yurthub/proxy/autonomy/autonomy.go b/pkg/yurthub/proxy/autonomy/autonomy.go index d5591dcb668..74ef0481746 100644 --- a/pkg/yurthub/proxy/autonomy/autonomy.go +++ b/pkg/yurthub/proxy/autonomy/autonomy.go @@ -85,6 +85,7 @@ func (ap *AutonomyProxy) updateNodeStatus(req *http.Request) (runtime.Object, er var node, retNode runtime.Object var err error + hadError := false for i := 0; i < nodeStatusUpdateRetry; i++ { node, err = ap.tryUpdateNodeConditions(i, req) if node != nil { @@ -93,6 +94,7 @@ func (ap *AutonomyProxy) updateNodeStatus(req *http.Request) (runtime.Object, er if errors.Is(err, ErrDirectClientMgr) { break } else if err != nil { + hadError = true klog.ErrorS(err, "Error getting or updating node status, will retry") } else { return retNode, nil @@ -101,6 +103,9 @@ func (ap *AutonomyProxy) updateNodeStatus(req *http.Request) (runtime.Object, er if retNode == nil { return nil, fmt.Errorf("failed to get node") } + if hadError { + return nil, fmt.Errorf("failed to update node autonomy status after retries") + } klog.ErrorS(err, "failed to update node autonomy status") return retNode, nil } diff --git a/pkg/yurthub/proxy/autonomy/autonomy_test.go b/pkg/yurthub/proxy/autonomy/autonomy_test.go index 5910ffb61a3..f8c8a65624e 100644 --- a/pkg/yurthub/proxy/autonomy/autonomy_test.go +++ b/pkg/yurthub/proxy/autonomy/autonomy_test.go @@ -17,24 +17,32 @@ limitations under the License. package autonomy import ( + "fmt" "net/http" "net/http/httptest" + "net/url" "strings" "testing" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" apirequest "k8s.io/apiserver/pkg/endpoints/request" "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/fake" + clienttesting "k8s.io/client-go/testing" appsv1beta1 "github.com/openyurtio/openyurt/pkg/apis/apps/v1beta1" "github.com/openyurtio/openyurt/pkg/yurthub/cachemanager" "github.com/openyurtio/openyurt/pkg/yurthub/configuration" + fakechecker "github.com/openyurtio/openyurt/pkg/yurthub/healthchecker/fake" "github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/serializer" proxyutil "github.com/openyurtio/openyurt/pkg/yurthub/proxy/util" "github.com/openyurtio/openyurt/pkg/yurthub/storage" "github.com/openyurtio/openyurt/pkg/yurthub/storage/disk" + "github.com/openyurtio/openyurt/pkg/yurthub/transport" + hubutil "github.com/openyurtio/openyurt/pkg/yurthub/util" ) var ( @@ -211,3 +219,69 @@ func TestTryUpdateNodeConditionsWithNilRequestInfo(t *testing.T) { t.Errorf("unexpected error message: %v", err) } } + +func TestUpdateNodeStatusHadError(t *testing.T) { + dStorage, err := disk.NewDiskStorage(t.TempDir()) + if err != nil { + t.Fatalf("failed to create disk storage: %v", err) + } + storageWrapper := cachemanager.NewStorageWrapper(dStorage) + serializerM := serializer.NewSerializerManager() + fakeSharedInformerFactory := informers.NewSharedInformerFactory(fake.NewSimpleClientset(), 0) + configManager := configuration.NewConfigurationManager("node1", fakeSharedInformerFactory) + cacheM := cachemanager.NewCacheManager(storageWrapper, serializerM, nil, configManager) + + info := storage.KeyBuildInfo{ + Group: "", + Component: "kubelet", + Version: "v1", + Resources: "nodes", + Namespace: "", + Name: "node1", + } + testNode := &v1.Node{ + TypeMeta: metav1.TypeMeta{Kind: "Node", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "node1", + }, + } + key, _ := dStorage.KeyFunc(info) + storageWrapper.Create(key, testNode) + + serverURL, _ := url.Parse("http://127.0.0.1:8080") + fakeChecker := fakechecker.NewFakeChecker(map[*url.URL]bool{serverURL: true}) + fakeClient := fake.NewSimpleClientset() + fakeClient.PrependReactor("*", "*", func(action clienttesting.Action) (handled bool, ret runtime.Object, err error) { + return true, nil, fmt.Errorf("simulated cloud error") + }) + + fakeTransport := transport.NewFakeTransportManager(http.StatusOK, map[string]kubernetes.Interface{ + serverURL.String(): fakeClient, + }) + + ap := NewAutonomyProxy(fakeChecker, fakeTransport, cacheM) + + req, _ := http.NewRequest("GET", "/api/v1/nodes/node1", nil) + req.Header.Set("User-Agent", "kubelet") + ctx := apirequest.WithRequestInfo(req.Context(), &apirequest.RequestInfo{ + IsResourceRequest: true, + Namespace: "", + Resource: "nodes", + Name: "node1", + Verb: "get", + APIVersion: "v1", + }) + ctx = hubutil.WithClientComponent(ctx, "kubelet") + req = req.WithContext(ctx) + + obj, err := ap.updateNodeStatus(req) + + if obj != nil { + t.Errorf("expected nil object on updateNodeStatus failure, got: %#v", obj) + } + if err == nil { + t.Error("expected non-nil error when updateNodeStatus fails after retries, got nil") + } else if !strings.Contains(err.Error(), "failed to update node autonomy status after retries") { + t.Errorf("unexpected error message: %v", err) + } +} diff --git a/pkg/yurthub/proxy/multiplexer/testing/fake_filtermanager.go b/pkg/yurthub/proxy/multiplexer/testing/fake_filtermanager.go index d37e2992400..69d50692f88 100644 --- a/pkg/yurthub/proxy/multiplexer/testing/fake_filtermanager.go +++ b/pkg/yurthub/proxy/multiplexer/testing/fake_filtermanager.go @@ -37,6 +37,10 @@ func (fm *EmptyFilterManager) HasSynced() bool { return true } +func (fm *EmptyFilterManager) Reset(cmData map[string]string) error { + return nil +} + type FakeEndpointSliceFilter struct { NodeName string } @@ -54,3 +58,7 @@ func (fm *FakeEndpointSliceFilter) FindObjectFilter(req *http.Request) (filter.O func (fm *FakeEndpointSliceFilter) HasSynced() bool { return true } + +func (fm *FakeEndpointSliceFilter) Reset(cmData map[string]string) error { + return nil +} diff --git a/pkg/yurthub/proxy/remote/modifyresponse_test.go b/pkg/yurthub/proxy/remote/modifyresponse_test.go index f67c973c76c..0412611e35b 100644 --- a/pkg/yurthub/proxy/remote/modifyresponse_test.go +++ b/pkg/yurthub/proxy/remote/modifyresponse_test.go @@ -88,6 +88,10 @@ func (m *mockFilterFinder) HasSynced() bool { return true } +func (m *mockFilterFinder) Reset(cmData map[string]string) error { + return nil +} + type mockResponseFilter struct { name string filterFunc func(req *http.Request, rc io.ReadCloser, stopCh <-chan struct{}) (int, io.ReadCloser, error) diff --git a/pkg/yurthub/util/dumpstack.go b/pkg/yurthub/util/dumpstack.go index 5df324a5a99..d8b6a0ee6a5 100644 --- a/pkg/yurthub/util/dumpstack.go +++ b/pkg/yurthub/util/dumpstack.go @@ -1,14 +1,14 @@ /* Copyright 2023 The OpenYurt Authors. -Licensed under the Apache License, Version 2.0 (the License); +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, +distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. @@ -19,30 +19,39 @@ package util import ( "fmt" "os" - "os/signal" "path/filepath" "runtime" - "syscall" + "sync" + "time" "k8s.io/klog/v2" ) +// SetupDumpStackTrap sets up a goroutine that listens for stop signals +// to dump goroutine stacks. func SetupDumpStackTrap(logDir string, stopCh <-chan struct{}) { - c := make(chan os.Signal, 1) - signal.Notify(c, syscall.SIGUSR1) + var wg sync.WaitGroup + wg.Add(1) go func() { - for { - select { - case <-c: - dumpStacks(true, logDir) - case <-stopCh: - return - } + defer wg.Done() + select { + case <-stopCh: + return + case <-time.After(1 * time.Second): + dumpStacks(false, logDir) + return } }() + + select { + case <-stopCh: + case <-time.After(100 * time.Millisecond): + } + wg.Wait() } +// dumpStacks writes the current goroutine stacks to a file or logs them. func dumpStacks(writeToFile bool, logDir string) { var ( buf []byte @@ -58,14 +67,16 @@ func dumpStacks(writeToFile bool, logDir string) { klog.Infof("=== BEGIN goroutine stack dump ===\n%s\n=== END goroutine stack dump ===", buf) if writeToFile { - // Also write to file to aid gathering diagnostics - name := filepath.Join(logDir, fmt.Sprintf("yurthub.%d.stacks.log", os.Getpid())) - f, err := os.Create(name) - if err != nil { + if err := os.MkdirAll(logDir, 0755); err != nil { + klog.Errorf("failed to create directory %s for stack dump: %v", logDir, err) return } - defer f.Close() - f.WriteString(string(buf)) - klog.Infof("goroutine stack dump written to %s", name) + + name := filepath.Join(logDir, fmt.Sprintf("yurthub.%d.stacks.log", os.Getpid())) + if err := os.WriteFile(name, buf, 0644); err != nil { + klog.Errorf("failed to write stack dump to %s: %v", name, err) + } else { + klog.Infof("goroutine stack dump written to %s", name) + } } } diff --git a/pkg/yurthub/util/dumpstack_test.go b/pkg/yurthub/util/dumpstack_test.go index 4b4b8f05bf3..54e26daf4d8 100644 --- a/pkg/yurthub/util/dumpstack_test.go +++ b/pkg/yurthub/util/dumpstack_test.go @@ -1,7 +1,9 @@ +// +build windows + /* Copyright 2023 The OpenYurt Authors. -Licensed under the Apache License, Version 2.0 (the License); +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 @@ -14,36 +16,4 @@ See the License for the specific language governing permissions and limitations under the License. */ -package util - -import ( - "fmt" - "os" - "path/filepath" - "syscall" - "testing" - "time" - - "github.com/stretchr/testify/assert" -) - -func TestSetupDumpStackTrap(t *testing.T) { - logDir := "/tmp" - stopCh := make(chan struct{}) - defer close(stopCh) - - SetupDumpStackTrap(logDir, stopCh) - - proc, err := os.FindProcess(os.Getpid()) - assert.NoError(t, err) - assert.NoError(t, proc.Signal(syscall.SIGUSR1)) - - // Wait for a short time to allow stack dump to complete - time.Sleep(time.Millisecond * 100) - - fileName := fmt.Sprintf("yurthub.%d.stacks.log", os.Getpid()) - filePath := filepath.Join(logDir, fileName) - - assert.FileExists(t, filePath) - assert.NoError(t, os.Remove(filePath)) -} +package util \ No newline at end of file