fix: apiserver watcher refresh shouldn't crash plugin manager (#576)

# Description

This PR modifies the APIServer watcher to improve error handling and
optimize performance:

1. **Error Handling**: Instead of returning a resolution error to the
caller, watcher now retries the host lookup, after 3 refresh, if we
still can't resolve the apiserver IP, we bail and return the error to
the watcher manager.
2. **Optimization**: Moved the hostname retrieval to the initialization
function. This change ensures that the hostname is obtained once during
initialization and not on every refresh, as the hostname does not
change.

## Related Issue

[#565](#565)

If this pull request is related to any issue, please mention it here.
Additionally, make sure that the issue is assigned to you before
submitting this pull request.

## Checklist

- [x] I have read the [contributing
documentation](https://retina.sh/docs/contributing).
- [x] I signed and signed-off the commits (`git commit -S -s ...`). See
[this
documentation](https://docs.github.com/en/authentication/managing-commit-signature-verification/about-commit-signature-verification)
on signing commits.
- [x] I have correctly attributed the author(s) of the code.
- [x] I have tested the changes locally.
- [x] I have followed the project's style guidelines.
- [x] I have updated the documentation, if necessary.
- [ ] I have added tests, if applicable.

## Screenshots (if applicable) or Testing Completed

Please add any relevant screenshots or GIFs to showcase the changes
made.

## Additional Notes

Add any additional notes or context about the pull request here.

---

Please refer to the [CONTRIBUTING.md](../CONTRIBUTING.md) file for more
information on how to contribute to this project.

Author: jimassa

pkg/managers/pluginmanager/pluginmanager_test.go

@@ -120,6 +120,7 @@ func TestNewManagerStart(t *testing.T) {
 
 	for _, tt := range tests {
 		mgr, err := NewPluginManager(tt.cfg, tel, api.PluginName(tt.pluginName))
+		mgr.watcherManager = setupWatcherManagerMock(gomock.NewController(t))
 		require.Nil(t, err, "Expected nil but got error:%w", err)
 		require.NotNil(t, mgr, "Expected mgr to be intialized but found nil")
 		require.Condition(t, assert.Comparison(func() bool {

pkg/managers/watchermanager/watchermanager.go

@@ -36,7 +36,7 @@ func (wm *WatcherManager) Start(ctx context.Context) error {
 
 	for _, w := range wm.Watchers {
 		if err := w.Init(ctx); err != nil {
-			wm.l.Error("init failed", zap.String("watcher_type", fmt.Sprintf("%T", w)))
+			wm.l.Error("init failed", zap.String("watcher_type", fmt.Sprintf("%T", w)), zap.Error(err))
 			return err
 		}
 		wm.wg.Add(1)

pkg/managers/watchermanager/watchermanager_test.go

@@ -14,12 +14,28 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+var errInitFailed = errors.New("init failed")
+
 func TestStopWatcherManagerGracefully(t *testing.T) {
 	ctl := gomock.NewController(t)
 	defer ctl.Finish()
 	log.SetupZapLogger(log.GetDefaultLogOpts())
 	mgr := NewWatcherManager()
 
+	mockAPIServerWatcher := mock.NewMockIWatcher(ctl)
+	mockEndpointWatcher := mock.NewMockIWatcher(ctl)
+
+	mgr.Watchers = []IWatcher{
+		mockEndpointWatcher,
+		mockAPIServerWatcher,
+	}
+
+	mockAPIServerWatcher.EXPECT().Init(gomock.Any()).Return(nil).AnyTimes()
+	mockEndpointWatcher.EXPECT().Init(gomock.Any()).Return(nil).AnyTimes()
+
+	mockEndpointWatcher.EXPECT().Stop(gomock.Any()).Return(nil).AnyTimes()
+	mockAPIServerWatcher.EXPECT().Stop(gomock.Any()).Return(nil).AnyTimes()
+
 	ctx, _ := context.WithCancel(context.Background())
 	g, errctx := errgroup.WithContext(ctx)
 
@@ -37,17 +53,17 @@ func TestWatcherInitFailsGracefully(t *testing.T) {
 	defer ctl.Finish()
 	log.SetupZapLogger(log.GetDefaultLogOpts())
 
-	mockApiServerWatcher := mock.NewMockIWatcher(ctl)
+	mockAPIServerWatcher := mock.NewMockIWatcher(ctl)
 	mockEndpointWatcher := mock.NewMockIWatcher(ctl)
 
 	mgr := NewWatcherManager()
 	mgr.Watchers = []IWatcher{
-		mockApiServerWatcher,
+		mockAPIServerWatcher,
 		mockEndpointWatcher,
 	}
 
-	mockApiServerWatcher.EXPECT().Init(gomock.Any()).Return(errors.New("error")).AnyTimes()
-	mockEndpointWatcher.EXPECT().Init(gomock.Any()).Return(errors.New("error")).AnyTimes()
+	mockAPIServerWatcher.EXPECT().Init(gomock.Any()).Return(errInitFailed).AnyTimes()
+	mockEndpointWatcher.EXPECT().Init(gomock.Any()).Return(errInitFailed).AnyTimes()
 
 	err := mgr.Start(context.Background())
 	require.NotNil(t, err, "Expected error when starting watcher manager")

pkg/watchers/apiserver/apiserver.go

@@ -15,34 +15,37 @@ import (
 	"github.com/microsoft/retina/pkg/log"
 	fm "github.com/microsoft/retina/pkg/managers/filtermanager"
 	"github.com/microsoft/retina/pkg/pubsub"
+	"github.com/microsoft/retina/pkg/utils"
 	"go.uber.org/zap"
+	"k8s.io/client-go/rest"
 	kcfg "sigs.k8s.io/controller-runtime/pkg/client/config"
 )
 
 const (
 	filterManagerRetries = 3
+	hostLookupRetries    = 6 // 6 retries for a total of 63 seconds.
 )
 
 type ApiServerWatcher struct {
-	isRunning     bool
-	l             *log.ZapLogger
-	current       cache
-	new           cache
-	apiServerUrl  string
-	hostResolver  IHostResolver
-	filtermanager fm.IFilterManager
+	isRunning         bool
+	l                 *log.ZapLogger
+	current           cache
+	new               cache
+	apiServerHostName string
+	hostResolver      IHostResolver
+	filterManager     fm.IFilterManager
+	restConfig        *rest.Config
 }
 
 var a *ApiServerWatcher
 
-// Watcher creates a new apiserver watcher.
+// Watcher creates a new ApiServerWatcher instance.
 func Watcher() *ApiServerWatcher {
 	if a == nil {
 		a = &ApiServerWatcher{
 			isRunning:    false,
 			l:            log.Logger().Named("apiserver-watcher"),
 			current:      make(cache),
-			apiServerUrl: getHostURL(),
 			hostResolver: net.DefaultResolver,
 		}
 	}
@@ -56,12 +59,39 @@ func (a *ApiServerWatcher) Init(ctx context.Context) error {
 		return nil
 	}
 
-	a.filtermanager = getFilterManager()
+	// Get filter manager.
+	if a.filterManager == nil {
+		var err error
+		a.filterManager, err = fm.Init(filterManagerRetries)
+		if err != nil {
+			a.l.Error("failed to init filter manager", zap.Error(err))
+			return fmt.Errorf("failed to init filter manager: %w", err)
+		}
+	}
+
+	// Get  kubeconfig.
+	if a.restConfig == nil {
+		config, err := kcfg.GetConfig()
+		if err != nil {
+			a.l.Error("failed to get kubeconfig", zap.Error(err))
+			return fmt.Errorf("failed to get kubeconfig: %w", err)
+		}
+		a.restConfig = config
+	}
+
+	hostName, err := a.getHostName()
+	if err != nil {
+		a.l.Error("failed to get host name", zap.Error(err))
+		return fmt.Errorf("failed to get host name: %w", err)
+	}
+	a.apiServerHostName = hostName
+
 	a.isRunning = true
+
 	return nil
 }
 
-// Stop the apiserver watcher.
+// Stop stops the ApiServerWatcher.
 func (a *ApiServerWatcher) Stop(ctx context.Context) error {
 	if !a.isRunning {
 		a.l.Info("apiserver watcher is not running")
@@ -74,61 +104,57 @@ func (a *ApiServerWatcher) Stop(ctx context.Context) error {
 func (a *ApiServerWatcher) Refresh(ctx context.Context) error {
 	err := a.initNewCache(ctx)
 	if err != nil {
+		a.l.Error("failed to initialize new cache", zap.Error(err))
 		return err
 	}
-	// Compare the new ips with the old ones.
+
+	// Compare the new IPs with the old ones.
 	created, deleted := a.diffCache()
 
-	// Publish the new ips.
-	createdIps := []net.IP{}
-	deletedIps := []net.IP{}
+	createdIPs := []net.IP{}
+	deletedIPs := []net.IP{}
 
 	for _, v := range created {
-		a.l.Info("New Apiserver ips:", zap.Any("ip", v))
+		a.l.Info("New Apiserver IPs:", zap.Any("ip", v))
 		ip := net.ParseIP(v.(string)).To4()
-		createdIps = append(createdIps, ip)
+		createdIPs = append(createdIPs, ip)
 	}
 
 	for _, v := range deleted {
-		a.l.Info("Deleted Apiserver ips:", zap.Any("ip", v))
+		a.l.Info("Deleted Apiserver IPs:", zap.Any("ip", v))
 		ip := net.ParseIP(v.(string)).To4()
-		deletedIps = append(deletedIps, ip)
+		deletedIPs = append(deletedIPs, ip)
 	}
 
-	if len(createdIps) > 0 {
-		// Publish the new ips.
-		a.publish(createdIps, cc.EventTypeAddAPIServerIPs)
-		// Add ips to filter manager if any.
-		err := a.filtermanager.AddIPs(createdIps, "apiserver-watcher", fm.RequestMetadata{RuleID: "apiserver-watcher"})
+	if len(createdIPs) > 0 {
+		a.publish(createdIPs, cc.EventTypeAddAPIServerIPs)
+		err := a.filterManager.AddIPs(createdIPs, "apiserver-watcher", fm.RequestMetadata{RuleID: "apiserver-watcher"})
 		if err != nil {
-			a.l.Error("Failed to add ips to filter manager", zap.Error(err))
+			a.l.Error("Failed to add IPs to filter manager", zap.Error(err))
 		}
 	}
 
-	if len(deletedIps) > 0 {
-		// Publish the deleted ips.
-		a.publish(deletedIps, cc.EventTypeDeleteAPIServerIPs)
-		// Delete ips from filter manager if any.
-		err := a.filtermanager.DeleteIPs(deletedIps, "apiserver-watcher", fm.RequestMetadata{RuleID: "apiserver-watcher"})
+	if len(deletedIPs) > 0 {
+		a.publish(deletedIPs, cc.EventTypeDeleteAPIServerIPs)
+		err := a.filterManager.DeleteIPs(deletedIPs, "apiserver-watcher", fm.RequestMetadata{RuleID: "apiserver-watcher"})
 		if err != nil {
-			a.l.Error("Failed to delete ips from filter manager", zap.Error(err))
+			a.l.Error("Failed to delete IPs from filter manager", zap.Error(err))
 		}
 	}
 
-	// update the current cache and reset the new cache
 	a.current = a.new.deepcopy()
 	a.new = nil
 
 	return nil
 }
 
 func (a *ApiServerWatcher) initNewCache(ctx context.Context) error {
-	ips, err := a.getApiServerIPs(ctx)
+	ips, err := a.resolveIPs(ctx, a.apiServerHostName)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to resolve IPs: %w", err)
 	}
 
-	// Reset the new cache.
+	// Reset new cache.
 	a.new = make(cache)
 	for _, ip := range ips {
 		a.new[ip] = struct{}{}
@@ -137,14 +163,14 @@ func (a *ApiServerWatcher) initNewCache(ctx context.Context) error {
 }
 
 func (a *ApiServerWatcher) diffCache() (created, deleted []interface{}) {
-	// check if there are new ips
+	// Check if there are any new IPs.
 	for k := range a.new {
 		if _, ok := a.current[k]; !ok {
 			created = append(created, k)
 		}
 	}
 
-	// check if there are deleted ips
+	// Check if there are any deleted IPs.
 	for k := range a.current {
 		if _, ok := a.new[k]; !ok {
 			deleted = append(deleted, k)
@@ -153,53 +179,35 @@ func (a *ApiServerWatcher) diffCache() (created, deleted []interface{}) {
 	return
 }
 
-func (a *ApiServerWatcher) getApiServerIPs(ctx context.Context) ([]string, error) {
-	// Parse the URL
-	host, err := a.retrieveApiServerHostname()
-	if err != nil {
-		return nil, err
-	}
-
-	// Get the ips for the host
-	ips, err := a.resolveIPs(ctx, host)
-	if err != nil {
-		return nil, err
-	}
-
-	return ips, nil
-}
-
-// parse url to extract hostname
-func (a *ApiServerWatcher) retrieveApiServerHostname() (string, error) {
-	// Parse the URL
-	url, err := url.Parse(a.apiServerUrl)
-	if err != nil {
-		fmt.Println("Failed to parse URL:", err)
-		return "", err
-	}
-
-	// Remove the scheme (http:// or https://) and port from the host
-	host := strings.TrimPrefix(url.Host, "www.")
-	colonIndex := strings.IndexByte(host, ':')
-	if colonIndex != -1 {
-		host = host[:colonIndex]
+func (a *ApiServerWatcher) resolveIPs(ctx context.Context, host string) ([]string, error) {
+	// perform a DNS lookup for the host URL using the net.DefaultResolver which uses the local resolver.
+	// Possible errors  here are:
+	// 	- Canceled context: The context was canceled before the lookup completed.
+	// 	-DNS server errors ie NXDOMAIN, SERVFAIL.
+	// 	- Network errors ie timeout, unreachable DNS server.
+	// 	-Other DNS-related errors encapsulated in a DNSError.
+	var hostIPs []string
+	var err error
+
+	retryFunc := func() error {
+		hostIPs, err = a.hostResolver.LookupHost(ctx, host)
+		if err != nil {
+			return fmt.Errorf("APIServer LookupHost failed: %w", err)
+		}
+		return nil
 	}
-	return host, nil
-}
 
-// Resolve the list of ips for the given host
-func (a *ApiServerWatcher) resolveIPs(ctx context.Context, host string) ([]string, error) {
-	hostIps, err := a.hostResolver.LookupHost(ctx, host)
+	// Retry the lookup for hostIPs in case of failure.
+	err = utils.Retry(retryFunc, hostLookupRetries)
 	if err != nil {
 		return nil, err
 	}
 
-	if len(hostIps) == 0 {
-		a.l.Error("no ips found for host", zap.String("host", host))
-		return nil, fmt.Errorf("no ips found for host %s", host)
+	if len(hostIPs) == 0 {
+		a.l.Debug("no IPs found for host", zap.String("host", host))
 	}
 
-	return hostIps, nil
+	return hostIPs, nil
 }
 
 func (a *ApiServerWatcher) publish(netIPs []net.IP, eventType cc.EventType) {
@@ -212,30 +220,23 @@ func (a *ApiServerWatcher) publish(netIPs []net.IP, eventType cc.EventType) {
 		ipsToPublish = append(ipsToPublish, ip.String())
 	}
 	ps := pubsub.New()
-	ps.Publish(common.PubSubAPIServer,
-		cc.NewCacheEvent(
-			eventType,
-			common.NewAPIServerObject(ipsToPublish),
-		),
-	)
+	ps.Publish(common.PubSubAPIServer, cc.NewCacheEvent(eventType, common.NewAPIServerObject(ipsToPublish)))
 	a.l.Debug("Published event", zap.Any("eventType", eventType), zap.Any("netIPs", ipsToPublish))
 }
 
-// getHostURL returns the host url from the config.
-func getHostURL() string {
-	config, err := kcfg.GetConfig()
+func (a *ApiServerWatcher) getHostName() (string, error) {
+	// Parse the host URL.
+	hostURL := a.restConfig.Host
+	parsedURL, err := url.ParseRequestURI(hostURL)
 	if err != nil {
-		log.Logger().Error("failed to get config", zap.Error(err))
-		return ""
+		log.Logger().Error("failed to parse URL", zap.String("url", hostURL), zap.Error(err))
+		return "", fmt.Errorf("failed to parse URL: %w", err)
 	}
-	return config.Host
-}
 
-// Get FilterManager
-func getFilterManager() *fm.FilterManager {
-	f, err := fm.Init(filterManagerRetries)
-	if err != nil {
-		a.l.Error("failed to init filter manager", zap.Error(err))
+	// Extract the host name from the URL.
+	host := strings.TrimPrefix(parsedURL.Host, "www.")
+	if colonIndex := strings.IndexByte(host, ':'); colonIndex != -1 {
+		host = host[:colonIndex]
 	}
-	return f
+	return host, nil
 }