Repository: skupperproject/skupper
Pull request #1359 (2024-01-16 19:35:36)

Adr 8


Patch:
diff --git a/client/router_update.go b/client/router_update.go
index 53437fb8..6ce0be06 100644
--- a/client/router_update.go
+++ b/client/router_update.go
@@ -103,6 +103,7 @@ func (cli *VanClient) RouterUpdateVersionInNamespace(ctx context.Context, hup bo
 	substituteFlowCollector := false
 	addPrometheusServer := false
 	moveClaims := false
+	updateRole := false
 	inprogress, originalVersion, err := cli.isUpdating(namespace)
 	if err != nil {
 		return false, err
@@ -116,7 +117,10 @@ func (cli *VanClient) RouterUpdateVersionInNamespace(ctx context.Context, hup bo
 		addCertsSharedVolume = utils.LessRecentThanVersion(originalVersion, "0.9.0")
 		substituteFlowCollector = utils.LessRecentThanVersion(originalVersion, "1.3.0")
 		addPrometheusServer = utils.LessRecentThanVersion(originalVersion, "1.4.0")
-		moveClaims = utils.LessRecentThanVersion(originalVersion, "1.5.0")
+		if utils.LessRecentThanVersion(originalVersion, "1.5.0") {
+			moveClaims = !config.IsEdge()
+			updateRole = true //config-sync requires extra permission to write skupper-network-status configmap
+		}
 	} else {
 		originalVersion = site.Version
 	}
@@ -141,7 +145,8 @@ func (cli *VanClient) RouterUpdateVersionInNamespace(ctx context.Context, hup bo
 				addPrometheusServer = true
 			}
 			if utils.LessRecentThanVersion(originalVersion, "1.5.0") {
-				moveClaims = true
+				moveClaims = !config.IsEdge()
+				updateRole = true //config-sync requires extra permission to write skupper-network-status configmap
 			}
 
 			err = cli.updateStarted(site.Version, namespace, configmap.ObjectMeta.OwnerReferences)
@@ -524,6 +529,11 @@ func (cli *VanClient) RouterUpdateVersionInNamespace(ctx context.Context, hup bo
 		}
 		kube.AppendSecretVolumeWithVolumeName(&router.Spec.Template.Spec.Volumes, &router.Spec.Template.Spec.Containers[1].VolumeMounts, types.SiteServerSecret, "claims-cert", "/etc/skupper-internal/")
 		updateRouter = true
+	} else if updateRole {
+		err = kube.UpdateRole(namespace, types.TransportRoleName, types.TransportPolicyRule, cli.KubeClient)
+		if err != nil {
+			return false, err
+		}
 	}
 
 	if updateRouter || updateSite || hup {
diff --git a/cmd/config-sync/collector.go b/cmd/config-sync/collector.go
index 9143bc6c..e375a4f4 100644
--- a/cmd/config-sync/collector.go
+++ b/cmd/config-sync/collector.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"log"
 	"os"
 	"time"
@@ -31,41 +32,62 @@ func updateLockOwner(lockname, namespace string, owner *metav1.OwnerReference, c
 	return err
 }
 
-func siteCollector(stopCh <-chan struct{}, cli *client.VanClient) *flow.FlowCollector {
+func siteCollector(stopCh <-chan struct{}, cli *client.VanClient) {
 	var fc *flow.FlowCollector
 	siteData := map[string]string{}
 	platform := config.GetPlatform()
-	if platform == "" || platform == types.PlatformKubernetes {
-		current, err := kube.GetDeployment(types.TransportDeploymentName, cli.Namespace, cli.KubeClient)
-		if err != nil {
-			log.Fatal("Failed to get transport deployment", err.Error())
-		}
-		owner := kube.GetDeploymentOwnerReference(current)
+	if platform != types.PlatformKubernetes {
+		return
+	}
+	current, err := kube.GetDeployment(types.TransportDeploymentName, cli.Namespace, cli.KubeClient)
+	if err != nil {
+		log.Fatal("Failed to get transport deployment", err.Error())
+	}
+	owner := kube.GetDeploymentOwnerReference(current)
 
-		existing, err := kube.NewConfigMap(types.NetworkStatusConfigMapName, &siteData, nil, nil, &owner, cli.Namespace, cli.KubeClient)
-		if err != nil && existing == nil {
-			log.Fatal("Failed to create site status config map ", err.Error())
-		}
+	existing, err := kube.NewConfigMap(types.NetworkStatusConfigMapName, &siteData, nil, nil, &owner, cli.Namespace, cli.KubeClient)
+	if err != nil && existing == nil {
+		log.Fatal("Failed to create site status config map ", err.Error())
+	}
 
-		err = updateLockOwner(types.SiteLeaderLockName, cli.Namespace, &owner, cli)
-		if err != nil {
-			log.Println("Update lock error", err.Error())
-		}
+	err = updateLockOwner(types.SiteLeaderLockName, cli.Namespace, &owner, cli)
+	if err != nil {
+		log.Println("Update lock error", err.Error())
+	}
 
-		fc = flow.NewFlowCollector(flow.FlowCollectorSpec{
-			Mode:              flow.RecordStatus,
-			Namespace:         cli.Namespace,
-			Origin:            os.Getenv("SKUPPER_SITE_ID"),
-			PromReg:           nil,
-			ConnectionFactory: qdr.NewConnectionFactory("amqp://localhost:5672", nil),
-			FlowRecordTtl:     time.Minute * 15})
+	fc = flow.NewFlowCollector(flow.FlowCollectorSpec{
+		Mode:              flow.RecordStatus,
+		Namespace:         cli.Namespace,
+		PromReg:           nil,
+		ConnectionFactory: qdr.NewConnectionFactory("amqp://localhost:5672", nil),
+		FlowRecordTtl:     time.Minute * 15})
+
+	go primeBeacons(fc, cli)
+	log.Println("COLLECTOR: Starting flow collector")
+	fc.Start(stopCh)
+}
 
-		fc.Start(stopCh)
+// primeBeacons attempts to guess the router and service-controller vanflow IDs
+// to pass to the flow collector in order to accelerate startup time.
+func primeBeacons(fc *flow.FlowCollector, cli *client.VanClient) {
+	podname, _ := os.Hostname()
+	var prospectRouterID string
+	if len(podname) >= 5 {
+		prospectRouterID = fmt.Sprintf("%s:0", podname[len(podname)-5:])
+	}
+	var siteID string
+	cm, err := kube.WaitConfigMapCreated(types.SiteConfigMapName, cli.Namespace, cli.KubeClient, 5*time.Second, 250*time.Millisecond)
+	if err != nil {
+		log.Printf("COLLECTOR: failed to get skupper-site ConfigMap. Proceeding without Site ID. %s\n", err)
+	} else if cm != nil {
+		siteID = string(cm.ObjectMeta.UID)
 	}
-	return fc
+	log.Printf("COLLECTOR: Priming site with expected beacons for '%s' and '%s'\n", prospectRouterID, siteID)
+	fc.PrimeSiteBeacons(siteID, prospectRouterID)
 }
 
 func runLeaderElection(lock *resourcelock.ConfigMapLock, ctx context.Context, id string, cli *client.VanClient) {
+	begin := time.Now()
 	var stopCh chan struct{}
 	podname, _ := os.Hostname()
 	leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
@@ -76,7 +98,7 @@ func runLeaderElection(lock *resourcelock.ConfigMapLock, ctx context.Context, id
 		RetryPeriod:     2 * time.Second,
 		Callbacks: leaderelection.LeaderCallbacks{
 			OnStartedLeading: func(c context.Context) {
-				log.Printf("COLLECTOR: Leader %s starting site collection \n", podname)
+				log.Printf("COLLECTOR: Leader %s starting site collection after %s\n", podname, time.Since(begin))
 				stopCh = make(chan struct{})
 				siteCollector(stopCh, cli)
 			},
diff --git a/cmd/config-sync/main.go b/cmd/config-sync/main.go
index a370968f..3bf444ea 100644
--- a/cmd/config-sync/main.go
+++ b/cmd/config-sync/main.go
@@ -72,6 +72,9 @@ func main() {
 		log.Fatal("Error waiting for router pods to be ready ", err.Error())
 	}
 
+	log.Println("CONFIG_SYNC: Starting collector...")
+	go startCollector(cli)
+
 	event.StartDefaultEventStore(stopCh)
 	if claims.StartClaimVerifier(cli.KubeClient, cli.Namespace, cli, cli) {
 		log.Println("CONFIG_SYNC: Claim verifier started")
@@ -98,8 +101,6 @@ func main() {
 	log.Println("CONFIG_SYNC: Starting sync controller loop...")
 	configSync.start(stopCh)
 
-	go startCollector(cli)
-
 	<-stopCh
 	log.Println("CONFIG_SYNC: Shutting down...")
 	configSync.stop()
diff --git a/cmd/flow-collector/handlers.go b/cmd/flow-collector/handlers.go
index 0591d1d1..07d480f1 100644
--- a/cmd/flow-collector/handlers.go
+++ b/cmd/flow-collector/handlers.go
@@ -176,12 +176,20 @@ func (c *Controller) promqueryHandler(w http.ResponseWriter, r *http.Request) {
 	proxyReq, err := http.NewRequest(r.Method, urlOut, nil)
 	if err != nil {
 		log.Printf("COLLECTOR: prom proxy request error: %s\n", err.Error())
+
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprintf(w, "Internal Server Error: %s\n", err.Error())
+		return
 	}
 
 	w.Header().Set("Content-Type", "application/json")
 	proxyResp, err := client.Do(proxyReq)
 	if err != nil {
 		log.Printf("COLLECTOR: Prometheus query error: %s\n", err.Error())
+
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprintf(w, "Internal Server Error: %s\n", err.Error())
+		return
 	} else {
 		w.WriteHeader(proxyResp.StatusCode)
 		data, _ := io.ReadAll(proxyResp.Body)
@@ -197,6 +205,10 @@ func (c *Controller) promqueryrangeHandler(w http.ResponseWriter, r *http.Reques
 	proxyReq, err := http.NewRequest(r.Method, urlOut, nil)
 	if err != nil {
 		log.Printf("COLLECTOR: prom proxy request error: %s \n", err.Error())
+
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprintf(w, "Internal Server Error: %s\n", err.Error())
+		return
 	}
 
 	w.Header().Set("Content-Type", "application/json")
@@ -204,6 +216,10 @@ func (c *Controller) promqueryrangeHandler(w http.ResponseWriter, r *http.Reques
 	w.WriteHeader(proxyResp.StatusCode)
 	if err != nil {
 		log.Printf("COLLECTOR: Prometheus query_range error: %s\n", err.Error())
+
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprintf(w, "Internal Server Error: %s\n", err.Error())
+		return
 	} else {
 		data, _ := io.ReadAll(proxyResp.Body)
 		proxyResp.Body.Close()
diff --git a/cmd/get/get.go b/cmd/get/get.go
index 693a3644..b0530e00 100644
--- a/cmd/get/get.go
+++ b/cmd/get/get.go
@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/skupperproject/skupper/pkg/cleanhttp"
 	"github.com/skupperproject/skupper/pkg/utils"
 	"github.com/spf13/cobra"
 )
@@ -15,18 +16,17 @@ import (
 func get(path string, output string) error {
 	var resp *http.Response
 	var err error
-
+	client := cleanhttp.DefaultClient()
+	client.Timeout = time.Second * 10
 	err = utils.Retry(time.Second, 30, func() (bool, error) {
 		url := "http://localhost:8181/" + path
 		if output == "json" {
 			url += "?output=json"
 		}
-		resp, err = http.Get(url)
+
+		resp, err = client.Get(url)
 		if err != nil {
-			if strings.Contains(err.Error(), "connect: connection refused") {
-				return false, nil
-			}
-			return true, err
+			return false, nil
 		}
 		return true, nil
 	})
diff --git a/cmd/service-controller/main.go b/cmd/service-controller/main.go
index 5a06712d..c2a3769e 100644
--- a/cmd/service-controller/main.go
+++ b/cmd/service-controller/main.go
@@ -77,7 +77,7 @@ func main() {
 	}
 
 	log.Println("Waiting for Skupper router component to start")
-	_, err = kube.WaitDeploymentReady(types.TransportDeploymentName, namespace, cli.KubeClient, time.Second*180, time.Second*5)
+	_, err = kube.WaitDeploymentReady(types.TransportDeploymentName, namespace, cli.KubeClient, time.Second*180, time.Second)
 	if err != nil {
 		log.Fatal("Error waiting for transport deployment to be ready: ", err.Error())
 	}
diff --git a/cmd/skupper/skupper.go b/cmd/skupper/skupper.go
index ee834f17..ad8d3fc7 100644
--- a/cmd/skupper/skupper.go
+++ b/cmd/skupper/skupper.go
@@ -1127,7 +1127,6 @@ func init() {
 	cmdNetwork.AddCommand(NewCmdNetworkStatus(skupperCli.Network()))
 
 	cmdSwitch := NewCmdSwitch()
-	cmdSwitch.Hidden = true
 
 	addCommands(skupperCli, rootCmd,
 		cmdInit,
diff --git a/cmd/skupper/skupper_kube_site.go b/cmd/skupper/skupper_kube_site.go
index 2c73e195..8d888093 100644
--- a/cmd/skupper/skupper_kube_site.go
+++ b/cmd/skupper/skupper_kube_site.go
@@ -341,7 +341,7 @@ func (s *SkupperKubeSite) Update(cmd *cobra.Command, args []string) error {
 		return err
 	}
 	if updated {
-		fmt.Println("Skupper is now updated in '" + cli.GetNamespace() + "'.")
+		fmt.Println("Skupper update in progress for '" + cli.GetNamespace() + "'.")
 	} else {
 		fmt.Println("No update required in '" + cli.GetNamespace() + "'.")
 	}
diff --git a/doc/adr/0006-explicit-resources-for-declaring-service-bindings.md b/doc/adr/0006-explicit-resources-for-declaring-service-bindings.md
new file mode 100644
index 00000000..85a5c334
--- /dev/null
+++ b/doc/adr/0006-explicit-resources-for-declaring-service-bindings.md
@@ -0,0 +1,73 @@
+# 6. Explicit resources for declaring service bindings
+
+Date: 2024-01-16
+
+## Status
+
+Proposed
+
+Enables [7. Deprecate ServiceSync](0007-deprecate-servicesync.md)
+
+## Context
+
+Service bindings are the configurations that determine how a Skupper
+service is invoked (ingress binding) and what processes that
+invocation may be routed to (egress binding).
+
+The Deployments or StatefulSets that are to be annotated may
+themselves be the generated output of other controllers. In this case
+it is not always possible to ensure that they are annotated correctly
+for Skupper.
+
+The way annotations work when applied to Service resources is
+confusing and unintuitive.
+
+If the skupper.io/proxy annotation is found on a Service, the egress
+binding is configured based on the original selector in the Service
+and this selector is then changed to point the service at the router
+pods. This changing of the original resource can cause problems when a
+GitOps approach is used for deployment.
+
+However if the skupper.io/target annotation is also applied to the
+Service, the behaviour is different. Here the annotated service is
+updated to point at the skupper router pods, and the value of target
+is used to configure the egress binding.
+
+If instead the skupper.io/ingress-only annotation is also present on
+the Service, no target is inferred. The service is configured to point
+at the router pods on a port that will route to the address specified
+by the skupper.io/address annotation.
+
+The code accummulated to handle all these configurations as well as
+the CLI is not the easiest to maintain.
+
+## Decision
+
+Explicit resources will be introduced that describe the desired
+configuration for both ingress- and egress- bindings.
+
+## Consequences
+
+Support for the existing annotations, such as skupper.io/proxy, will
+be removed. It may be separately decided to provide legacy support in
+some fashion that would derive the new explicit bindings from
+annotated resources.
+
+It is expected that a model made explicit in the resources used will
+be easier for users to understand and internalise, making it more
+likely that they are successful in their use of skupper and reducing
+their need to seek help with simple configuration.
+
+Explicit bindings resources will also give more flexibility, for
+example allowing more than one ingress binding for the same service if
+needed, or allowing the hostname used for the ingress to vary by site.
+
+However a change to the model for service bindings will make upgrade
+and backward compatibility more of a challenge.
+
+The change in model will need to be clearly documented and explained
+to users, along with the recommended path for migration to this new
+model.
+
+The change would also impact existing examples, both under the
+skupperproject organisation and elsewhere.
diff --git a/doc/adr/0007-deprecate-servicesync.md b/doc/adr/0007-deprecate-servicesync.md
new file mode 100644
index 00000000..d4323241
--- /dev/null
+++ b/doc/adr/0007-deprecate-servicesync.md
@@ -0,0 +1,51 @@
+# 7. Deprecate ServiceSync
+
+Date: 2024-01-16
+
+## Status
+
+Proposed
+
+Enabled By [6. Explicit resources for declaring service bindings](0006-explicit-resources-for-declaring-service-bindings.md)
+
+## Context
+
+ServiceSync is a mechanism whereby service ingress bindings are
+automatically created at each site in the network for any service with
+egress bindings.
+
+Though this feature seemed nice in early demos, in practice it is
+generally unnecessary. In some cases it is explicitly not desirable as
+the ingress bindings should only be created in particular sites.
+
+Though the feature can at present be turned off, the option to do so
+is not obvious. Further it is not well tested.
+
+If the decision to have explicit resources for ingress bindings is
+adopted, it becomes even simpler to explicitly add the required
+ingresses to any site that requires them. Indeed as that proposal
+allows more flexibility with regard to the ingress bindings
+ServiceSync would not be adequate as it is.
+
+## Decision
+
+ServiceSync will be removed for automatic propoagation of ingress
+bindings between sites that have migrated to use explicit resources
+for each ingress binding.
+
+## Consequences
+
+This will require extra declarations to be made by user at the sites
+at which they want to consume services. However this also gives much
+more flexibility not only in the location of the ingress bindings, but
+also in their configuration.
+
+Upgrading a network that currently relies on ServiceSync may be more
+awkward if this decision is accepted. It would be possible to
+alleviate some of that by for example providing a legacy support
+process that would turn the new model into the expected sequences for
+the old ServiceSync protocol. What level of effort is appropriate for
+easing upgrade will be decided through separate ADRs.
+
+This change would also impact existing examples, both under the
+skupperproject organisation and elsewhere.
diff --git a/doc/adr/0008-combine-site-controller-and-service-controller.md b/doc/adr/0008-combine-site-controller-and-service-controller.md
new file mode 100644
index 00000000..f537dea7
--- /dev/null
+++ b/doc/adr/0008-combine-site-controller-and-service-controller.md
@@ -0,0 +1,62 @@
+# 8. Combine site-controller and service-controller
+
+Date: 2024-01-16
+
+## Status
+
+Proposed
+
+## Context
+
+The site-controller was introduced as a way of allowing declarative
+configuration of skupper for a namespace in Kubernetes. It watched for
+the skupper-site ConfigMap and then used essentially the same code as
+was used by the CLI to initialise the namespace. A new separate
+process was the least disruptive way to support declarative use at
+that time.
+
+There have been two changes that make it easier to revisit this
+configuration.
+
+The first is the removal of the original console which was tied to the
+service-controller. The flow-collector and the current console can be
+deployed quite independent of the service-controller now.
+
+The second is the introduction of the config-sync container and the
+implementation of a leader election among instances if there are more
+than one. This can be used to relocate any code that requires direct
+connection to the router.
+
+## Decision
+
+A new controller will be created that will combine the functions of
+the current site- and service- controllers. This new controller will
+be able to run either watching all namespaces or watching the current
+namespace, similar to the current site-controller.
+
+The controller will also support updating the site in line with any
+configuration changes.
+
+Features that require connection to the router will be moved into the
+config-sync container.
+
+## Consequences
+
+Removing an unnecessary component means simpler configuration and
+simpler code for automating that configuration. With this change the
+controller is concerned only with the configuration of the data
+plane. The configuration of the controller itself can be achieved by
+customing the yaml (as is currently the case for the site-controller).
+
+Skupper will also use one less pod per site in many cases.
+
+The get command and associated web api will no longer be available in
+its current form. Neither will the already deprecated web api for
+managing links and tokens. The prefered approach for making
+information available will be through kubernetes resources. An
+alternative to `skupper debug events` will be needed. This might be
+Kubernetes Event resources or some other resource into which
+information can be written.
+
+Upgrade from the current configuration to this new one will need some
+consideration.
diff --git a/pkg/cleanhttp/cleanhttp.go b/pkg/cleanhttp/cleanhttp.go
new file mode 100644
index 00000000..90b29d35
--- /dev/null
+++ b/pkg/cleanhttp/cleanhttp.go
@@ -0,0 +1,37 @@
+// cleanhttp has convenience functions for creating clean http clients and
+// transports - free of the globally mutable state in http.DefaultClient and
+// http.DefaultTransport variables.
+package cleanhttp
+
+import (
+	"net"
+	"net/http"
+	"time"
+)
+
+// DefaultClient returns a clean http Client using the same defaults as go's
+// http.DefaultClient without relying on global state shared with other
+// clients.
+func DefaultClient() *http.Client {
+	return &http.Client{
+		Transport: DefaultTransport(),
+	}
+}
+
+// DefaultTransport returns a clean http Transport with the same defaults as
+// go's http.DefaultTransport.
+func DefaultTransport() *http.Transport {
+	transport := &http.Transport{
+		Proxy: http.ProxyFromEnvironment,
+		DialContext: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).DialContext,
+		ForceAttemptHTTP2:     true,
+		MaxIdleConns:          100,
+		IdleConnTimeout:       90 * time.Second,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+	}
+	return transport
+}
diff --git a/pkg/flow/collector.go b/pkg/flow/collector.go
index 436c196c..ce6b1c54 100644
--- a/pkg/flow/collector.go
+++ b/pkg/flow/collector.go
@@ -3,6 +3,7 @@ package flow
 import (
 	"bytes"
 	"encoding/gob"
+	"fmt"
 	"log"
 	"net/http"
 	"time"
@@ -197,6 +198,9 @@ type FlowCollector struct {
 	connectorsToReconcile   map[string]string
 	processesToReconcile    map[string]*ProcessRecord
 	aggregatesToReconcile   map[string]*FlowPairRecord
+
+	begin           time.Time
+	networkStatusUp bool
 }
 
 func getTtl(ttl time.Duration) time.Duration {
@@ -332,7 +336,7 @@ func (c *FlowCollector) beaconUpdate(beacon BeaconRecord) {
 }
 
 func (c *FlowCollector) recordUpdates(stopCh <-chan struct{}) {
-	tickerFlush := time.NewTicker(1 * time.Second)
+	tickerFlush := time.NewTicker(250 * time.Millisecond)
 	defer tickerFlush.Stop()
 	tickerReconcile := time.NewTicker(2 * time.Second)
 	defer tickerReconcile.Stop()
@@ -398,9 +402,39 @@ func (c *FlowCollector) recordUpdates(stopCh <-chan struct{}) {
 }
 
 func (c *FlowCollector) Start(stopCh <-chan struct{}) {
+	c.begin = time.Now()
 	go c.run(stopCh)
 }
 
+// PrimeSiteBeacons "sends" the collector phony beacon records for a controller
+// and a router event source when their addresses are known in order to avoid
+// the startup time waiting for beacon records.
+func (c *FlowCollector) PrimeSiteBeacons(controllerID, routerID string) {
+	var incoming []interface{}
+	if controllerID != "" {
+		incoming = append(incoming, BeaconRecord{
+			Version:    1,
+			SourceType: "CONTROLLER",
+			Address:    fmt.Sprintf("mc/sfe.%s", controllerID),
+			Direct:     fmt.Sprintf("sfe.%s", controllerID),
+			Identity:   controllerID,
+		})
+	}
+	if routerID != "" {
+		incoming = append(incoming, BeaconRecord{
+			Version:    1,
+			SourceType: "ROUTER",
+			Address:    fmt.Sprintf("mc/sfe.%s", routerID),
+			Direct:     fmt.Sprintf("sfe.%s", routerID),
+			Identity:   routerID,
+		})
+	}
+	if len(incoming) == 0 {
+		return
+	}
+	c.beaconsIncoming <- incoming
+}
+
 func (c *FlowCollector) run(stopCh <-chan struct{}) {
 	if c.mode == RecordMetrics {
 		c.metrics = c.NewMetrics(c.prometheusReg)
diff --git a/pkg/flow/flow_mem_driver.go b/pkg/flow/flow_mem_driver.go
index 4cee897b..b5b3b7f2 100644
--- a/pkg/flow/flow_mem_driver.go
+++ b/pkg/flow/flow_mem_driver.go
@@ -487,6 +487,8 @@ var defaultRetry = wait.Backoff{
 	Jitter:   0.1,
 }
 
+var netUpdateCt int
+
 func (fc *FlowCollector) updateNetworkStatus() error {
 	var err error
 	networkData := map[string]string{}
@@ -545,9 +547,14 @@ func (fc *FlowCollector) updateNetworkStatus() error {
 			if err != nil {
 				return err
 			} else {
+				netUpdateCt++
 				return nil
 			}
 		})
+		if !fc.networkStatusUp && len(networkStatus.Sites) > 0 && len(networkStatus.Sites[0].RouterStatus) > 0 {
+			fc.networkStatusUp = true
+			log.Printf("COLLECTOR: First functional network status update written after %s and %d updates\n", time.Since(fc.begin), netUpdateCt)
+		}
 	}
 	return err
 }
diff --git a/pkg/flow/record.go b/pkg/flow/record.go
index 5691058d..cb8879ab 100644
--- a/pkg/flow/record.go
+++ b/pkg/flow/record.go
@@ -474,15 +474,15 @@ type Payload struct {
 }
 
 type QueryParams struct {
-	Offset             int               `json:"offset"`
-	Limit              int               `json:"limit"`
-	SortBy             string            `json:"sortBy"`
-	Filter             string            `json:"filter"`
-	FilterFields       map[string]string `json:"filterFields"`
-	TimeRangeStart     uint64            `json:"timeRangeStart"`
-	TimeRangeEnd       uint64            `json:"timeRangeEnd"`
-	TimeRangeOperation TimeRangeRelation `json:"timeRangeOperation"`
-	State              RecordState       `json:"state"`
+	Offset             int                 `json:"offset"`
+	Limit              int                 `json:"limit"`
+	SortBy             string              `json:"sortBy"`
+	Filter             string              `json:"filter"`
+	FilterFields       map[string][]string `json:"filterFields"`
+	TimeRangeStart     uint64              `json:"timeRangeStart"`
+	TimeRangeEnd       uint64              `json:"timeRangeEnd"`
+	TimeRangeOperation TimeRangeRelation   `json:"timeRangeOperation"`
+	State              RecordState         `json:"state"`
 }
 
 func getQueryParams(url *url.URL) QueryParams {
@@ -492,7 +492,7 @@ func getQueryParams(url *url.URL) QueryParams {
 		Limit:              -1,
 		SortBy:             "identity.asc",
 		Filter:             "",
-		FilterFields:       make(map[string]string),
+		FilterFields:       make(map[string][]string),
 		TimeRangeStart:     now - (15 * oneMinute),
 		TimeRangeEnd:       now,
 		TimeRangeOperation: intersects,
@@ -556,7 +556,7 @@ func getQueryParams(url *url.URL) QueryParams {
 				qp.State = all
 			}
 		default:
-			qp.FilterFields[cases.Title(language.Und, cases.NoLower).String(k)] = v[0]
+			qp.FilterFields[cases.Title(language.Und, cases.NoLower).String(k)] = v
 		}
 	}
 	return qp
@@ -700,33 +700,37 @@ func getField(field string, record interface{}) interface{} {
 	return nil
 }
 
-func matchFieldValue(x interface{}, y string) bool {
-	if x != nil && y != "" {
-		switch x.(type) {
-		case string:
-			return strings.HasPrefix(x.(string), y)
-		case uint64:
-			i, err := strconv.ParseInt(y, 10, 64)
-			if err == nil {
-				return x.(uint64) == uint64(i)
-			}
-		case int32:
-			i, err := strconv.ParseInt(y, 10, 32)
-			if err == nil {
-				return x.(int32) == int32(i)
-			}
-		case int64:
-			i, err := strconv.ParseInt(y, 10, 64)
-			if err == nil {
-				return x.(int64) == int64(i)
+func matchFieldValues(x interface{}, values []string) bool {
+	if x == nil || len(values) == 0 {
+		return false
+	}
+	switch x := x.(type) {
+	case string:
+		for _, y := range values {
+			if y == "" {
+				continue
 			}
-		case int:
-			i, err := strconv.ParseInt(y, 10, 64)
-			if err == nil {
-				return x.(int) == int(i)
+			if strings.HasPrefix(x, y) {
+				return true
 			}
-		default:
-			return false
+		}
+	case uint64:
+		return numInStringSlice(x, values)
+	case int32:
+		return numInStringSlice(x, values)
+	case int64:
+		return numInStringSlice(x, values)
+	case int:
+		return numInStringSlice(x, values)
+	}
+	return false
+}
+
+func numInStringSlice[T int | uint64 | int64 | int32](x T, values []string) bool {
+	for _, value := range values {
+		i, err := strconv.ParseInt(value, 10, 64)
+		if err == nil && x == T(i) {
+			return true
 		}
 	}
 	return false
@@ -773,9 +777,9 @@ func compareFields(x, y interface{}, order string) bool {
 	}
 }
 
-// type any = interface{}
-func filterRecord[T any](item T, qp QueryParams) bool {
+func filterRecord(item any, qp QueryParams) bool {
 	filter := true
+	//TODO(ck) deprecate filter?
 	if qp.Filter != "" {
 		field, match, err := validateAndReturnFilterQuery(qp.Filter)
 		// todo propagate error or log
@@ -791,7 +795,7 @@ func filterRecord[T any](item T, qp QueryParams) bool {
 				if !filterSubRecord(value, match) {
 					filter = false
 				}
-			} else if !matchFieldValue(value, match) {
+			} else if !matchFieldValues(value, []string{match}) {
 				filter = false
 			}
 		}
@@ -807,10 +811,10 @@ func filterRecord[T any](item T, qp QueryParams) bool {
 			} else {
 				x := reflect.ValueOf(value)
 				if x.Kind() == reflect.Struct {
-					if !filterSubRecord(value, subField+"."+match) {
+					if !filterSubRecordValues(value, subField, match) {
 						filter = false
 					}
-				} else if !matchFieldValue(value, match) {
+				} else if !matchFieldValues(value, match) {
 					filter = false
 				}
 			}
@@ -819,8 +823,22 @@ func filterRecord[T any](item T, qp QueryParams) bool {
 	return filter
 }
 
-// type any = interface{}
-func filterSubRecord[T any](item T, filter string) bool {
+func filterSubRecordValues(item interface{}, spec string, values []string) bool {
+	if spec == "" {
+		return true
+	}
+	titleCaser := cases.Title(language.Und, cases.NoLower)
+	parts := strings.Split(spec, ".")
+	field, spec := titleCaser.String(parts[0]), strings.Join(parts[1:], ".")
+	value := getField(field, item)
+	x := reflect.ValueOf(value)
+	if x.Kind() == reflect.Struct {
+		return filterSubRecordValues(value, spec, values)
+	}
+	return matchFieldValues(value, values)
+}
+
+func filterSubRecord(item interface{}, filter string) bool {
 	if filter == "" {
 		return true
 	}
@@ -834,7 +852,7 @@ func filterSubRecord[T any](item T, filter string) bool {
 	if x.Kind() == reflect.Struct {
 		return filterSubRecord(value, match)
 	}
-	return matchFieldValue(value, match)
+	return matchFieldValues(value, []string{match})
 }
 
 func sortAndSlice[T any](list []T, payload *Payload, queryParams QueryParams) error {


Test patch:
diff --git a/pkg/flow/record_test.go b/pkg/flow/record_test.go
index 9564f970..31e18c22 100644
--- a/pkg/flow/record_test.go
+++ b/pkg/flow/record_test.go
@@ -1,6 +1,7 @@
 package flow
 
 import (
+	"fmt"
 	"net/http"
 	"testing"
 
@@ -359,13 +360,13 @@ func TestRecordState(t *testing.T) {
 
 func TestParameters(t *testing.T) {
 	type test struct {
-		values      map[string]string
+		url         string
 		queryParams QueryParams
 	}
 
 	testTable := []test{
 		{
-			values: map[string]string{"timeRangeStart": "1234", "timeRangeEnd": "0"},
+			url: "http://host?timeRangeStart=1234&timeRangeEnd=0",
 			queryParams: QueryParams{
 				Offset:             -1,
 				Limit:              -1,
@@ -377,7 +378,7 @@ func TestParameters(t *testing.T) {
 			},
 		},
 		{
-			values: map[string]string{"timeRangeStart": "0", "timeRangeEnd": "0", "timeRangeOperation": "contains"},
+			url: "http://host?timeRangeStart=0&timeRangeEnd=0&timeRangeOperation=contains",
 			queryParams: QueryParams{
 				Offset:             -1,
 				Limit:              -1,
@@ -389,7 +390,7 @@ func TestParameters(t *testing.T) {
 			},
 		},
 		{
-			values: map[string]string{"timeRangeStart": "1234", "timeRangeEnd": "0", "timeRangeOperation": "within"},
+			url: "http://host?timeRangeStart=1234&timeRangeEnd=0&timeRangeOperation=within",
 			queryParams: QueryParams{
 				Offset:             -1,
 				Limit:              -1,
@@ -401,15 +402,7 @@ func TestParameters(t *testing.T) {
 			},
 		},
 		{
-			values: map[string]string{
-				"offset":             "10",
-				"limit":              "10",
-				"sortBy":             "sourcePort.desc",
-				"filter":             "forwardFlow.protocol.tcp",
-				"timeRangeStart":     "0",
-				"timeRangeEnd":       "4567",
-				"timeRangeOperation": "intersects",
-			},
+			url: "http://host?timeRangeStart=0&timeRangeEnd=4567&timeRangeOperation=intersects&offset=10&limit=10&sortBy=sourcePort.desc&filter=forwardFlow.protocol.tcp",
 			queryParams: QueryParams{
 				Offset:             10,
 				Limit:              10,
@@ -420,22 +413,30 @@ func TestParameters(t *testing.T) {
 				TimeRangeOperation: intersects,
 			},
 		},
+		{
+			url: "http://host?processRole=external&processRole=internal&timeRangeStart=0&timeRangeEnd=0&limit=0&offset=0",
+			queryParams: QueryParams{
+				SortBy: "identity.asc",
+				FilterFields: map[string][]string{
+					"processRole": {"external", "internal"},
+				},
+			},
+		},
 	}
 
 	for _, test := range testTable {
-		req, _ := http.NewRequest("GET", "/", nil)
-		q := req.URL.Query()
-		for k, v := range test.values {
-			q.Add(k, v)
-		}
-		req.URL.RawQuery = q.Encode()
-		qp := getQueryParams(req.URL)
-		assert.Equal(t, qp.Offset, test.queryParams.Offset)
-		assert.Equal(t, qp.Limit, test.queryParams.Limit)
-		assert.Equal(t, qp.SortBy, test.queryParams.SortBy)
-		assert.Equal(t, qp.TimeRangeStart, test.queryParams.TimeRangeStart)
-		assert.Equal(t, qp.TimeRangeEnd, test.queryParams.TimeRangeEnd)
-		assert.Equal(t, qp.TimeRangeOperation, test.queryParams.TimeRangeOperation)
+		t.Run(test.url, func(t *testing.T) {
+			req, _ := http.NewRequest("GET", test.url, nil)
+			q := req.URL.Query()
+			req.URL.RawQuery = q.Encode()
+			qp := getQueryParams(req.URL)
+			assert.Equal(t, qp.Offset, test.queryParams.Offset)
+			assert.Equal(t, qp.Limit, test.queryParams.Limit)
+			assert.Equal(t, qp.SortBy, test.queryParams.SortBy)
+			assert.Equal(t, qp.TimeRangeStart, test.queryParams.TimeRangeStart)
+			assert.Equal(t, qp.TimeRangeEnd, test.queryParams.TimeRangeEnd)
+			assert.Equal(t, qp.TimeRangeOperation, test.queryParams.TimeRangeOperation)
+		})
 	}
 }
 
@@ -488,27 +489,27 @@ func TestPagination(t *testing.T) {
 
 func TestMatchField(t *testing.T) {
 	field1 := "foo"
-	field1Value := "foo"
+	field1Value := []string{"foo"}
 	field2 := uint64(12345678)
-	field2Value := "12345678"
+	field2Value := []string{"12345678"}
 	field3 := int32(87654321)
-	field3Value := "87654321"
+	field3Value := []string{"87654321"}
 	field4 := int64(12345678)
-	field4Value := "12345678"
+	field4Value := []string{"12345678"}
 	field5 := int(12345678)
-	field5Value := "12345678"
+	field5Value := []string{"12345678"}
 
-	match := matchFieldValue(field1, field1Value)
+	match := matchFieldValues(field1, field1Value)
 	assert.Equal(t, match, true)
-	match = matchFieldValue(field2, field2Value)
+	match = matchFieldValues(field2, field2Value)
 	assert.Equal(t, match, true)
-	match = matchFieldValue(field3, field3Value)
+	match = matchFieldValues(field3, field3Value)
 	assert.Equal(t, match, true)
-	match = matchFieldValue(field4, field4Value)
+	match = matchFieldValues(field4, field4Value)
 	assert.Equal(t, match, true)
-	match = matchFieldValue(field5, field5Value)
+	match = matchFieldValues(field5, field5Value)
 	assert.Equal(t, match, true)
-	match = matchFieldValue(field5, field1Value)
+	match = matchFieldValues(field5, field1Value)
 	assert.Equal(t, match, false)
 }
 
@@ -664,52 +665,66 @@ func TestFilterFieldsRecord(t *testing.T) {
 	}
 
 	type test struct {
-		filterField map[string]string
+		filterField map[string][]string
 		result      bool
 	}
 	testTable := []test{
 		{
-			filterField: map[string]string{"forwardFlow.SourceHost": "10.20.30.40"},
+			filterField: map[string][]string{"forwardFlow.SourceHost": {"10.20.30.40"}},
+			result:      true,
+		},
+		{
+			filterField: map[string][]string{"sourceSiteName": {"public1"}},
 			result:      true,
 		},
 		{
-			filterField: map[string]string{"sourceSiteName": "public1"},
+			filterField: map[string][]string{"sourceSiteName": {"pub"}},
 			result:      true,
 		},
 		{
-			filterField: map[string]string{"sourceSiteName": "pub"},
+			filterField: map[string][]string{"sourceSiteName": {"pub", "xyz"}},
 			result:      true,
 		},
 		{
-			filterField: map[string]string{"sourceSiteName": "ic1"},
+			filterField: map[string][]string{"sourceSiteName": {"ic1"}},
 			result:      false,
 		},
 		{
-			filterField: map[string]string{"sourceSiteName": "xpublic1"},
+			filterField: map[string][]string{"sourceSiteName": {"xpublic1"}},
 			result:      false,
 		},
 		{
-			filterField: map[string]string{"forwardFlow.SourceHost": "10.20.30.40", "sourceSiteName": "public1"},
+			filterField: map[string][]string{"sourceSiteName": {"xpublic1", "zpublic1"}},
+			result:      false,
+		},
+		{
+			filterField: map[string][]string{"forwardFlow.SourceHost": {"10.20.30.40"}, "sourceSiteName": {"public1"}},
 			result:      true,
 		},
 		{
-			filterField: map[string]string{"forwardFlow.SourceHost": "10.20.30.40", "sourceSiteName": "public2"},
+			filterField: map[string][]string{"forwardFlow.SourceHost": {"10.20.255.255", "10.20.30.40"}, "sourceSiteName": {"public1"}},
+			result:      true,
+		},
+		{
+			filterField: map[string][]string{"forwardFlow.SourceHost": {"10.20.30.40"}, "sourceSiteName": {"public2"}},
 			result:      false,
 		},
 		{
-			filterField: map[string]string{},
+			filterField: map[string][]string{},
 			result:      true,
 		},
 		{
-			filterField: map[string]string{"sourceSiteName": ""},
+			filterField: map[string][]string{"sourceSiteName": {""}},
 			result:      false,
 		},
 	}
 
-	for _, test := range testTable {
-		qp := QueryParams{FilterFields: test.filterField}
-		result := filterRecord(flowPair, qp)
-		assert.Equal(t, result, test.result)
+	for i, test := range testTable {
+		t.Run(fmt.Sprintf("test-%d", i), func(t *testing.T) {
+			qp := QueryParams{FilterFields: test.filterField}
+			result := filterRecord(flowPair, qp)
+			assert.Equal(t, result, test.result)
+		})
 	}
 }