watch.go

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

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