watch.go

Documentation: edge-infra.dev/pkg/edge/bsl

     1  package bsl
     2  
     3  import (
     4  	"context"
     5  	"reflect"
     6  
     7  	"github.com/go-logr/logr"
     8  
     9  	v1 "k8s.io/api/core/v1"
    10  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    11  	"k8s.io/apimachinery/pkg/fields"
    12  	kwatch "k8s.io/apimachinery/pkg/watch"
    13  
    14  	"sigs.k8s.io/controller-runtime/pkg/client"
    15  )
    16  
    17  // OnConfigMapUpdate watches for configmap updates, execute function based on optional filters
    18  // returns a watch ref that can be called to stop the watch `defer watch.Stop()`
    19  func OnConfigMapUpdate(ctx context.Context, cl client.WithWatch, logger logr.Logger, fn func(cfg *BSLInfo), filters ...string) (kwatch.Interface, error) {
    20  	list := &v1.ConfigMapList{}
    21  	watch, err := cl.Watch(ctx, list, &client.ListOptions{
    22  		FieldSelector: fields.OneTermEqualSelector("metadata.name", BSLInfoConfigMapName),
    23  		Namespace:     metav1.NamespacePublic,
    24  	})
    25  	if err != nil {
    26  		return nil, err
    27  	}
    28  	go func() {
    29  		var old *v1.ConfigMap
    30  		for event := range watch.ResultChan() {
    31  			switch event.Type {
    32  			case kwatch.Added, kwatch.Modified:
    33  				updatedConfigmap := event.Object.(*v1.ConfigMap)
    34  				edgeInfoUpdated, err := FromConfigMap(updatedConfigmap)
    35  				if err != nil { // skip for invalid config
    36  					logger.Error(err, "fail to parse bsl info ConfigMap, will watch for update")
    37  					continue
    38  				}
    39  				if old == nil {
    40  					old = updatedConfigmap
    41  					fn(edgeInfoUpdated)
    42  					continue
    43  				}
    44  				previousBslInfo, _ := FromConfigMap(old) // ignore err should already be valid
    45  				configMapChanged := !reflect.DeepEqual(previousBslInfo, edgeInfoUpdated)
    46  				matchedOnFilters := cfgValuesMatches(old, updatedConfigmap, filters)
    47  				if configMapChanged && matchedOnFilters {
    48  					old = updatedConfigmap
    49  					fn(edgeInfoUpdated)
    50  				}
    51  			}
    52  		}
    53  	}()
    54  	return watch, err
    55  }
    56  
    57  func cfgValuesMatches(old, updated *v1.ConfigMap, filters []string) bool {
    58  	for _, f := range filters {
    59  		if old.Data[f] != updated.Data[f] {
    60  			return true
    61  		}
    62  	}
    63  	return len(filters) == 0
    64  }
    65
View as plain text