Repository: skupperproject/skupper
Pull request #1096 (2023-04-18 16:45:24)

add multi field filtering


Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index e69b701e..1d09e837 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -528,6 +528,8 @@ jobs:
                 tar -zcf "${BASEDIR}/archives/skupper-cli-${VERSION}-$p.tgz" *
               fi
             done
+            make -e generate-manifest
+            cp ${BASEDIR}/manifest.json ${BASEDIR}/archives
             cd ${BASEDIR}
             ghr -t ${GITHUB_TOKEN} -u ${CIRCLE_PROJECT_USERNAME} -r ${CIRCLE_PROJECT_REPONAME} -c ${CIRCLE_SHA1} -replace -prerelease -draft ${VERSION} "${BASEDIR}/archives"
 
diff --git a/.gitignore b/.gitignore
index cf964a2a..fd6c9d9a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,6 +10,7 @@
 /get
 /config-sync
 /flow-collector
+/manifest
 
 # Test binary, built with `go test -c`
 *.test
@@ -25,3 +26,6 @@
 
 # Dependency directories (remove the comment below to include it)
 # vendor/
+
+# Generated files
+manifest.json
\ No newline at end of file
diff --git a/Makefile b/Makefile
index fe9bc0c1..8a1c0aa1 100644
--- a/Makefile
+++ b/Makefile
@@ -8,7 +8,7 @@ TEST_BINARIES_FOLDER := ${PWD}/test/integration/bin
 DOCKER := docker
 LDFLAGS := -X github.com/skupperproject/skupper/pkg/version.Version=${VERSION}
 
-all: build-cmd build-get build-config-sync build-controllers build-tests
+all: build-cmd build-get build-config-sync build-controllers build-tests build-manifest
 
 build-tests:
 	mkdir -p ${TEST_BINARIES_FOLDER}
@@ -39,6 +39,9 @@ build-config-sync:
 
 build-controllers: build-site-controller build-service-controller build-flow-collector
 
+build-manifest:
+	go build -ldflags="${LDFLAGS}"  -o manifest ./cmd/manifest
+
 docker-build-test-image:
 	${DOCKER} build -t ${TEST_IMAGE} -f Dockerfile.ci-test .
 
@@ -88,7 +91,7 @@ test:
 	go test -v -count=1 ./pkg/... ./cmd/... ./client/...
 
 clean:
-	rm -rf skupper service-controller site-controller release get config-sync ${TEST_BINARIES_FOLDER}
+	rm -rf skupper service-controller site-controller release get config-sync manifest ${TEST_BINARIES_FOLDER}
 
 package: release/windows.zip release/darwin.zip release/linux.tgz
 
@@ -109,3 +112,6 @@ release/darwin/skupper: cmd/skupper/skupper.go
 
 release/darwin.zip: release/darwin/skupper
 	zip -j release/darwin.zip release/darwin/skupper
+
+generate-manifest: build-manifest
+	./manifest
diff --git a/client/container/client.go b/client/container/client.go
index a8247dfe..621b5550 100644
--- a/client/container/client.go
+++ b/client/container/client.go
@@ -7,6 +7,7 @@ import (
 	"os"
 	"path"
 	"strings"
+	"time"
 
 	"github.com/skupperproject/skupper/api/types"
 )
@@ -68,9 +69,9 @@ type Container struct {
 	RestartPolicy string
 	RestartCount  int
 	Running       bool
-	CreatedAt     string
-	StartedAt     string
-	ExitedAt      string
+	CreatedAt     time.Time
+	StartedAt     time.Time
+	ExitedAt      time.Time
 	ExitCode      int
 }
 
diff --git a/client/container/informer.go b/client/container/informer.go
new file mode 100644
index 00000000..bd92c16e
--- /dev/null
+++ b/client/container/informer.go
@@ -0,0 +1,25 @@
+package container
+
+type Informer[T any] interface {
+	OnAdd(obj T)
+	OnUpdate(oldObj, newObj T)
+	OnDelete(obj T)
+}
+
+type InformerBase[T any] struct {
+	Add    func(obj T)
+	Update func(oldObj, newObj T)
+	Delete func(obj T)
+}
+
+func (e *InformerBase[T]) OnAdd(obj T) {
+	e.Add(obj)
+}
+
+func (e *InformerBase[T]) OnUpdate(oldObj, newObj T) {
+	e.Update(oldObj, newObj)
+}
+
+func (e *InformerBase[T]) OnDelete(obj T) {
+	e.Delete(obj)
+}
diff --git a/client/podman/container.go b/client/podman/container.go
index f6feb8af..c48a7e55 100644
--- a/client/podman/container.go
+++ b/client/podman/container.go
@@ -5,6 +5,7 @@ import (
 	"os/user"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/go-openapi/runtime"
 	"github.com/skupperproject/skupper/api/types"
@@ -243,18 +244,21 @@ func (p *PodmanRestClient) ContainerExec(id string, command []string) (string, e
 
 func FromListContainer(c models.ListContainer) *container.Container {
 	ct := &container.Container{
-		ID:        c.ID,
-		Name:      c.Names[0],
-		Pod:       c.Pod,
-		Image:     c.Image,
-		Labels:    c.Labels,
-		Command:   c.Command,
-		Running:   !c.Exited,
-		CreatedAt: fmt.Sprint(c.CreatedAt),
-		StartedAt: fmt.Sprint(c.StartedAt),
-		ExitedAt:  fmt.Sprint(c.ExitedAt),
-		ExitCode:  int(c.ExitCode),
-	}
+		ID:       c.ID,
+		Name:     c.Names[0],
+		Pod:      c.Pod,
+		Image:    c.Image,
+		Labels:   c.Labels,
+		Command:  c.Command,
+		Running:  !c.Exited,
+		ExitCode: int(c.ExitCode),
+	}
+	// when listing containers dates are returned in unix time
+	createdAt, _ := strconv.ParseInt(c.CreatedAt, 10, 64)
+	ct.CreatedAt = time.Unix(createdAt, 0)
+	ct.StartedAt = time.Unix(c.StartedAt, 0)
+	ct.ExitedAt = time.Unix(c.ExitedAt, 0)
+
 	ct.Networks = map[string]container.ContainerNetworkInfo{}
 	ct.Env = map[string]string{}
 
@@ -285,7 +289,7 @@ func FromInspectContainer(c containers.ContainerInspectLibpodOKBody) *container.
 		ID:           c.ID,
 		Name:         c.Name,
 		RestartCount: int(c.RestartCount),
-		CreatedAt:    c.Created.String(),
+		CreatedAt:    time.Time(c.Created),
 		Pod:          c.Pod,
 	}
 	ct.Networks = map[string]container.ContainerNetworkInfo{}
@@ -372,8 +376,8 @@ func FromInspectContainer(c containers.ContainerInspectLibpodOKBody) *container.
 	// State info
 	if c.State != nil {
 		ct.Running = c.State.Running
-		ct.StartedAt = c.State.StartedAt.String()
-		ct.ExitedAt = c.State.FinishedAt.String()
+		ct.StartedAt = time.Time(c.State.StartedAt)
+		ct.ExitedAt = time.Time(c.State.FinishedAt)
 		ct.ExitCode = int(c.State.ExitCode)
 	}
 
diff --git a/client/podman/informer.go b/client/podman/informer.go
new file mode 100644
index 00000000..c54aff2b
--- /dev/null
+++ b/client/podman/informer.go
@@ -0,0 +1,124 @@
+package podman
+
+import (
+	"log"
+	"time"
+
+	"github.com/skupperproject/skupper/client/container"
+	"github.com/skupperproject/skupper/pkg/utils"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+const (
+	resyncPeriod = time.Second * 5
+)
+
+//
+// Base types
+//
+
+type informerCommon struct {
+	cli          *PodmanRestClient
+	resyncPeriod time.Duration
+}
+
+func (i *informerCommon) SetResyncPeriod(t time.Duration) {
+	i.resyncPeriod = t
+}
+
+type updateNotification[T any] struct {
+	oldObj T
+	newObj T
+}
+
+//
+// Container informer
+//
+
+func NewContainerInformer(cli *PodmanRestClient) *ContainerInformer {
+	return &ContainerInformer{
+		informerCommon: informerCommon{
+			cli:          cli,
+			resyncPeriod: resyncPeriod,
+		},
+		containers: map[string]*container.Container{},
+		informers:  []container.Informer[*container.Container]{},
+	}
+}
+
+type ContainerInformer struct {
+	informerCommon
+	containers map[string]*container.Container
+	informers  []container.Informer[*container.Container]
+}
+
+func (c *ContainerInformer) AddInformer(i container.Informer[*container.Container]) {
+	c.informers = append(c.informers, i)
+}
+
+func (c *ContainerInformer) Start(stopCh chan struct{}) {
+	go wait.Until(c.run, c.resyncPeriod, stopCh)
+}
+
+func (c *ContainerInformer) run() {
+	cl, err := c.cli.ContainerList()
+	var clNames []string
+	if err != nil {
+		log.Printf("unable to retrieve container list - %s", err)
+		return
+	}
+
+	var added []*container.Container
+	var updated []updateNotification[*container.Container]
+	var deleted []*container.Container
+
+	// Verifying new and updated containers
+	for _, ci := range cl {
+		clNames = append(clNames, ci.Name)
+		if co, ok := c.containers[ci.Name]; !ok {
+			cn, err := c.cli.ContainerInspect(ci.Name)
+			if err != nil {
+				log.Printf("error inspecting container: %s - %s", ci.Name, err)
+				continue
+			}
+			c.containers[ci.Name] = cn
+			added = append(added, cn)
+		} else if co.ID != ci.ID || ci.StartedAt.After(co.StartedAt) {
+			cn, err := c.cli.ContainerInspect(ci.Name)
+			if err != nil {
+				log.Printf("error inspecting container: %s - %s", ci.Name, err)
+				continue
+			}
+			updated = append(updated, updateNotification[*container.Container]{
+				oldObj: co,
+				newObj: cn,
+			})
+			c.containers[ci.Name] = cn
+		}
+	}
+
+	// Verifying deleted containers
+	for ciName, ci := range c.containers {
+		if !utils.StringSliceContains(clNames, ciName) {
+			deleted = append(deleted, ci)
+			delete(c.containers, ciName)
+		}
+	}
+
+	// Notifying informers
+	for _, ci := range added {
+		for _, i := range c.informers {
+			i.OnAdd(ci)
+		}
+	}
+	for _, ci := range updated {
+		for _, i := range c.informers {
+			i.OnUpdate(ci.oldObj, ci.newObj)
+		}
+	}
+	for _, ci := range deleted {
+		for _, i := range c.informers {
+			i.OnDelete(ci)
+		}
+	}
+}
diff --git a/cmd/manifest/manifest.go b/cmd/manifest/manifest.go
new file mode 100644
index 00000000..9897cdf8
--- /dev/null
+++ b/cmd/manifest/manifest.go
@@ -0,0 +1,86 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/skupperproject/skupper/pkg/images"
+	"github.com/spf13/cobra"
+	"os"
+)
+
+type SkupperImage struct {
+	Name       string `yaml:"name"`
+	Repository string `yaml:"repository,omitempty"`
+}
+
+func generateManifestFile() error {
+
+	// Define an array of images.
+	skupperImages := []SkupperImage{
+		{
+			Name:       images.GetRouterImageName(),
+			Repository: "https://github.com/skupperproject/skupper-router",
+		},
+		{
+			Name:       images.GetServiceControllerImageName(),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetConfigSyncImageName(),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetFlowCollectorImageName(),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name: images.GetPrometheusServerImageName(),
+		},
+	}
+
+	// Define a struct for the manifest file.
+	manifest := struct {
+		Images []SkupperImage `json:"images"`
+	}{
+		Images: skupperImages,
+	}
+
+	// Encode the manifest image list as JSON.
+	manifestListJSON, err := json.MarshalIndent(manifest, "", "   ")
+	if err != nil {
+		return fmt.Errorf("Error encoding manifest image list: %v\n", err)
+
+	}
+
+	// Create a new file.
+	file, err := os.Create("manifest.json")
+	if err != nil {
+		fmt.Errorf("Error creating file: %v\n", err)
+	}
+
+	// Write the JSON data to the file.
+	_, err = file.Write(manifestListJSON)
+	if err != nil {
+		return fmt.Errorf("Error writing to file: %v\n", err)
+	}
+
+	return nil
+}
+
+func main() {
+	var command = &cobra.Command{
+		Use:   "manifest",
+		Short: "generates a manifest.json file",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return generateManifestFile()
+		},
+	}
+
+	command.Execute()
+
+	if err := command.Execute(); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}
diff --git a/pkg/flow/collector.go b/pkg/flow/collector.go
index 5a9389aa..5288cfcb 100644
--- a/pkg/flow/collector.go
+++ b/pkg/flow/collector.go
@@ -51,6 +51,7 @@ type collectorMetrics struct {
 	lastAccessed    *prometheus.GaugeVec
 	flowLatency     *prometheus.HistogramVec
 	activeReconcile *prometheus.GaugeVec
+	apiQueryLatency *prometheus.HistogramVec
 }
 
 func (fc *FlowCollector) NewMetrics(reg prometheus.Registerer) *collectorMetrics {
@@ -116,6 +117,14 @@ func (fc *FlowCollector) NewMetrics(reg prometheus.Registerer) *collectorMetrics
 				Help: "Number of active reconcile tasks, partitione by type",
 			},
 			[]string{"reconcileTask"}),
+		apiQueryLatency: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Name: "api_query_latency_microseconds",
+				Help: "The measure latency for the direction of query to api",
+				//                 10us,100us, 1ms,  2 ms, 5ms,  10ms,  100ms,  1s,      10s
+				Buckets: []float64{10, 100, 1000, 2000, 5000, 10000, 100000, 1000000, 10000000},
+			},
+			[]string{"recordType", "handler"}),
 	}
 	reg.MustRegister(m.info)
 	reg.MustRegister(m.collectorOctets)
@@ -127,6 +136,7 @@ func (fc *FlowCollector) NewMetrics(reg prometheus.Registerer) *collectorMetrics
 	reg.MustRegister(m.lastAccessed)
 	reg.MustRegister(m.flowLatency)
 	reg.MustRegister(m.activeReconcile)
+	reg.MustRegister(m.apiQueryLatency)
 	return m
 
 }
diff --git a/pkg/flow/flow_mem_driver.go b/pkg/flow/flow_mem_driver.go
index 128437e0..695b36b3 100644
--- a/pkg/flow/flow_mem_driver.go
+++ b/pkg/flow/flow_mem_driver.go
@@ -920,11 +920,11 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		Results:        nil,
 		QueryParams:    queryParams,
 		Status:         "",
-		Timestamp:      uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
-		Elapsed:        0,
 		Count:          0,
 		TimeRangeCount: 0,
 		TotalCount:     0,
+		timestamp:      uint64(time.Now().UnixNano()) / uint64(time.Microsecond),
+		elapsed:        0,
 	}
 
 	switch request.RecordType {
@@ -933,7 +933,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			sites := []SiteRecord{}
 			for _, site := range fc.Sites {
-				if filterRecord(*site, p.QueryParams.Filter) && site.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*site, p.QueryParams) && site.Base.TimeRangeValid(queryParams) {
 					sites = append(sites, *site)
 				}
 			}
@@ -953,7 +953,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, process := range fc.Processes {
 						if process.Parent == site.Identity {
 							p.TotalCount++
-							if filterRecord(*process, p.QueryParams.Filter) && process.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*process, p.QueryParams) && process.Base.TimeRangeValid(queryParams) {
 								processes = append(processes, *process)
 							}
 						}
@@ -968,7 +968,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, router := range fc.Routers {
 						if router.Parent == site.Identity {
 							p.TotalCount++
-							if filterRecord(*router, p.QueryParams.Filter) && router.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*router, p.QueryParams) && router.Base.TimeRangeValid(queryParams) {
 								routers = append(routers, *router)
 							}
 						}
@@ -983,7 +983,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, link := range fc.Links {
 						if fc.getRecordSiteId(*link) == site.Identity {
 							p.TotalCount++
-							if filterRecord(*link, p.QueryParams.Filter) && link.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*link, p.QueryParams) && link.Base.TimeRangeValid(queryParams) {
 								links = append(links, *link)
 							}
 						}
@@ -998,7 +998,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, host := range fc.Hosts {
 						if host.Parent == site.Identity {
 							p.TotalCount++
-							if filterRecord(*host, p.QueryParams.Filter) && host.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*host, p.QueryParams) && host.Base.TimeRangeValid(queryParams) {
 								hosts = append(hosts, *host)
 							}
 						}
@@ -1012,7 +1012,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			hosts := []HostRecord{}
 			for _, host := range fc.Hosts {
-				if filterRecord(*host, p.QueryParams.Filter) && host.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*host, p.QueryParams) && host.Base.TimeRangeValid(queryParams) {
 					hosts = append(hosts, *host)
 				}
 			}
@@ -1031,7 +1031,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			routers := []RouterRecord{}
 			for _, router := range fc.Routers {
-				if filterRecord(*router, p.QueryParams.Filter) && router.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*router, p.QueryParams) && router.Base.TimeRangeValid(queryParams) {
 					routers = append(routers, *router)
 				}
 			}
@@ -1053,13 +1053,13 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 							for _, flow := range fc.Flows {
 								if flow.Parent == connId {
 									p.TotalCount++
-									if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+									if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 										flows = append(flows, *flow)
 									}
 								} else if l4Flow, ok := fc.Flows[flow.Parent]; ok {
 									if l4Flow.Parent == connId {
 										p.TotalCount++
-										if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+										if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 											flows = append(flows, *flow)
 										}
 									}
@@ -1072,13 +1072,13 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 							for _, flow := range fc.Flows {
 								if flow.Parent == listenerId {
 									p.TotalCount++
-									if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+									if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 										flows = append(flows, *flow)
 									}
 								} else if l4Flow, ok := fc.Flows[flow.Parent]; ok {
 									if l4Flow.Parent == listenerId {
 										p.TotalCount++
-										if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+										if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 											flows = append(flows, *flow)
 										}
 									}
@@ -1096,7 +1096,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, link := range fc.Links {
 						if link.Parent == router.Identity {
 							p.TotalCount++
-							if filterRecord(*link, p.QueryParams.Filter) && link.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*link, p.QueryParams) && link.Base.TimeRangeValid(queryParams) {
 								links = append(links, *link)
 							}
 						}
@@ -1111,7 +1111,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, listener := range fc.Listeners {
 						if listener.Parent == router.Identity {
 							p.TotalCount++
-							if filterRecord(*listener, p.QueryParams.Filter) && listener.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*listener, p.QueryParams) && listener.Base.TimeRangeValid(queryParams) {
 								listeners = append(listeners, *listener)
 							}
 						}
@@ -1126,7 +1126,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, connector := range fc.Connectors {
 						if connector.Parent == router.Identity {
 							p.TotalCount++
-							if filterRecord(*connector, p.QueryParams.Filter) && connector.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*connector, p.QueryParams) && connector.Base.TimeRangeValid(queryParams) {
 								connectors = append(connectors, *connector)
 							}
 						}
@@ -1140,7 +1140,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			links := []LinkRecord{}
 			for _, link := range fc.Links {
-				if filterRecord(*link, p.QueryParams.Filter) && link.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*link, p.QueryParams) && link.Base.TimeRangeValid(queryParams) {
 					links = append(links, *link)
 				}
 			}
@@ -1159,7 +1159,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			listeners := []ListenerRecord{}
 			for _, listener := range fc.Listeners {
-				if filterRecord(*listener, p.QueryParams.Filter) && listener.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*listener, p.QueryParams) && listener.Base.TimeRangeValid(queryParams) {
 					listeners = append(listeners, *listener)
 				}
 			}
@@ -1179,13 +1179,13 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, flow := range fc.Flows {
 						if flow.Parent == id {
 							p.TotalCount++
-							if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 								flows = append(flows, *flow)
 							}
 						} else if l4Flow, ok := fc.Flows[flow.Parent]; ok {
 							if l4Flow.Parent == id {
 								p.TotalCount++
-								if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+								if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 									flows = append(flows, *flow)
 								}
 							}
@@ -1200,7 +1200,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			connectors := []ConnectorRecord{}
 			for _, connector := range fc.Connectors {
-				if filterRecord(*connector, p.QueryParams.Filter) && connector.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*connector, p.QueryParams) && connector.Base.TimeRangeValid(queryParams) {
 					connectors = append(connectors, *connector)
 				}
 			}
@@ -1220,13 +1220,13 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, flow := range fc.Flows {
 						if flow.Parent == id {
 							p.TotalCount++
-							if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 								flows = append(flows, *flow)
 							}
 						} else if l4Flow, ok := fc.Flows[flow.Parent]; ok {
 							if l4Flow.Parent == id {
 								p.TotalCount++
-								if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+								if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 									flows = append(flows, *flow)
 								}
 							}
@@ -1252,7 +1252,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			addresses := []VanAddressRecord{}
 			for _, address := range fc.VanAddresses {
-				if filterRecord(*address, p.QueryParams.Filter) {
+				if filterRecord(*address, p.QueryParams) {
 					fc.getAddressMetrics(address)
 					addresses = append(addresses, *address)
 				}
@@ -1276,13 +1276,13 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 							for _, flow := range fc.Flows {
 								if flow.Parent == connId && *connector.Protocol == "tcp" {
 									p.TotalCount++
-									if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+									if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 										flows = append(flows, *flow)
 									}
 								} else if l4Flow, ok := fc.Flows[flow.Parent]; ok {
 									if l4Flow.Parent == connId {
 										p.TotalCount++
-										if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+										if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 											flows = append(flows, *flow)
 										}
 									}
@@ -1295,13 +1295,13 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 							for _, flow := range fc.Flows {
 								if flow.Parent == listenerId && *listener.Protocol == "tcp" {
 									p.TotalCount++
-									if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+									if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 										flows = append(flows, *flow)
 									}
 								} else if l4Flow, ok := fc.Flows[flow.Parent]; ok {
 									if l4Flow.Parent == listenerId {
 										p.TotalCount++
-										if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+										if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 											flows = append(flows, *flow)
 										}
 									}
@@ -1323,7 +1323,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 								if flow.Parent == listenerId && flow.CounterFlow != nil {
 									if flowpair, ok := fc.FlowPairs["fp-"+flowId]; ok {
 										p.TotalCount++
-										if filterRecord(*flowpair, p.QueryParams.Filter) && flowpair.Base.TimeRangeValid(queryParams) {
+										if filterRecord(*flowpair, p.QueryParams) && flowpair.Base.TimeRangeValid(queryParams) {
 											flowPairs = append(flowPairs, *flowpair)
 										}
 									}
@@ -1331,7 +1331,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 									if l4Flow.Parent == listenerId {
 										if flowpair, ok := fc.FlowPairs["fp-"+flowId]; ok {
 											p.TotalCount++
-											if filterRecord(*flowpair, p.QueryParams.Filter) && flowpair.Base.TimeRangeValid(queryParams) {
+											if filterRecord(*flowpair, p.QueryParams) && flowpair.Base.TimeRangeValid(queryParams) {
 												flowPairs = append(flowPairs, *flowpair)
 											}
 										}
@@ -1351,7 +1351,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, connector := range fc.Connectors {
 						if *connector.Address == vanaddr.Name && connector.process != nil {
 							if process, ok := fc.Processes[*connector.process]; ok {
-								if filterRecord(*process, p.QueryParams.Filter) && process.Base.TimeRangeValid(queryParams) {
+								if filterRecord(*process, p.QueryParams) && process.Base.TimeRangeValid(queryParams) {
 									unique[process.Identity] = process
 								}
 							}
@@ -1371,7 +1371,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, listener := range fc.Listeners {
 						if *listener.Address == vanaddr.Name {
 							p.TotalCount++
-							if filterRecord(*listener, p.QueryParams.Filter) && listener.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*listener, p.QueryParams) && listener.Base.TimeRangeValid(queryParams) {
 								listeners = append(listeners, *listener)
 							}
 						}
@@ -1386,7 +1386,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, connector := range fc.Connectors {
 						if *connector.Address == vanaddr.Name {
 							p.TotalCount++
-							if filterRecord(*connector, p.QueryParams.Filter) && connector.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*connector, p.QueryParams) && connector.Base.TimeRangeValid(queryParams) {
 								connectors = append(connectors, *connector)
 							}
 						}
@@ -1400,7 +1400,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			processes := []ProcessRecord{}
 			for _, process := range fc.Processes {
-				if filterRecord(*process, p.QueryParams.Filter) && process.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*process, p.QueryParams) && process.Base.TimeRangeValid(queryParams) {
 					processes = append(processes, *process)
 				}
 			}
@@ -1419,7 +1419,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 				for _, flow := range fc.Flows {
 					if flow.Process != nil && *flow.Process == id {
 						p.TotalCount++
-						if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+						if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 							flows = append(flows, *flow)
 						}
 					}
@@ -1434,7 +1434,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						if connector.process != nil && *connector.process == id {
 							for _, address := range fc.VanAddresses {
 								if *connector.Address == address.Name {
-									if filterRecord(*address, p.QueryParams.Filter) {
+									if filterRecord(*address, p.QueryParams) {
 										fc.getAddressMetrics(address)
 										addresses = append(addresses, *address)
 									}
@@ -1464,7 +1464,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 			processGroups := []ProcessGroupRecord{}
 			for _, processGroup := range fc.ProcessGroups {
 				p.TotalCount++
-				if filterRecord(*processGroup, p.QueryParams.Filter) && processGroup.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*processGroup, p.QueryParams) && processGroup.Base.TimeRangeValid(queryParams) {
 					processGroups = append(processGroups, *processGroup)
 				}
 			}
@@ -1483,7 +1483,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 					for _, process := range fc.Processes {
 						if *process.GroupIdentity == processGroup.Identity {
 							p.TotalCount++
-							if filterRecord(*process, p.QueryParams.Filter) && process.Base.TimeRangeValid(queryParams) {
+							if filterRecord(*process, p.QueryParams) && process.Base.TimeRangeValid(queryParams) {
 								processes = append(processes, *process)
 							}
 						}
@@ -1497,7 +1497,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			flows := []FlowRecord{}
 			for _, flow := range fc.Flows {
-				if filterRecord(*flow, p.QueryParams.Filter) && flow.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*flow, p.QueryParams) && flow.Base.TimeRangeValid(queryParams) {
 					flows = append(flows, *flow)
 				}
 			}
@@ -1527,7 +1527,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			flowPairs := []FlowPairRecord{}
 			for _, flowPair := range fc.FlowPairs {
-				if filterRecord(*flowPair, p.QueryParams.Filter) && flowPair.Base.TimeRangeValid(queryParams) {
+				if filterRecord(*flowPair, p.QueryParams) && flowPair.Base.TimeRangeValid(queryParams) {
 					flowPairs = append(flowPairs, *flowPair)
 				}
 			}
@@ -1554,7 +1554,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						sourceId == *aggregate.SourceId && destinationId == "" ||
 						sourceId == "" && destinationId == *aggregate.DestinationId ||
 						sourceId == *aggregate.SourceId && destinationId == *aggregate.DestinationId {
-						if filterRecord(*aggregate, p.QueryParams.Filter) {
+						if filterRecord(*aggregate, p.QueryParams) {
 							aggregates = append(aggregates, *aggregate)
 						}
 					}
@@ -1584,7 +1584,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						sourceId == *aggregate.SourceId && destinationId == "" ||
 						sourceId == "" && destinationId == *aggregate.DestinationId ||
 						sourceId == *aggregate.SourceId && destinationId == *aggregate.DestinationId {
-						if filterRecord(*aggregate, p.QueryParams.Filter) {
+						if filterRecord(*aggregate, p.QueryParams) {
 							aggregates = append(aggregates, *aggregate)
 						}
 					}
@@ -1614,7 +1614,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 						sourceId == *aggregate.SourceId && destinationId == "" ||
 						sourceId == "" && destinationId == *aggregate.DestinationId ||
 						sourceId == *aggregate.SourceId && destinationId == *aggregate.DestinationId {
-						if filterRecord(*aggregate, p.QueryParams.Filter) {
+						if filterRecord(*aggregate, p.QueryParams) {
 							aggregates = append(aggregates, *aggregate)
 						}
 					}
@@ -1710,7 +1710,7 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 		case "list":
 			eventSources := []EventSourceRecord{}
 			for _, eventSource := range fc.eventSources {
-				if filterRecord(eventSource.EventSourceRecord, p.QueryParams.Filter) && eventSource.EventSourceRecord.Base.TimeRangeValid(queryParams) {
+				if filterRecord(eventSource.EventSourceRecord, p.QueryParams) && eventSource.EventSourceRecord.Base.TimeRangeValid(queryParams) {
 					eventSources = append(eventSources, eventSource.EventSourceRecord)
 				}
 			}
@@ -1742,7 +1742,11 @@ func (fc *FlowCollector) retrieve(request ApiRequest) (*string, error) {
 	if retrieveError != nil {
 		p.Status = retrieveError.Error()
 	}
-	p.Elapsed = uint64(time.Now().UnixNano())/uint64(time.Microsecond) - p.Timestamp
+	p.elapsed = uint64(time.Now().UnixNano())/uint64(time.Microsecond) - p.timestamp
+	apiQueryLatencyMetric, err := fc.metrics.apiQueryLatency.GetMetricWith(map[string]string{"recordType": recordNames[request.RecordType], "handler": request.HandlerName})
+	if err == nil {
+		apiQueryLatencyMetric.Observe(float64(p.elapsed))
+	}
 	data, err := json.MarshalIndent(p, "", " ")
 	if err != nil {
 		log.Println("Error marshalling results", err.Error())
diff --git a/pkg/flow/record.go b/pkg/flow/record.go
index bdd36741..6192a956 100644
--- a/pkg/flow/record.go
+++ b/pkg/flow/record.go
@@ -418,11 +418,11 @@ type Payload struct {
 	Results        interface{} `json:"results"`
 	QueryParams    QueryParams `json:"queryParams"`
 	Status         string      `json:"status"`
-	Timestamp      uint64      `json:"timestamp"`
-	Elapsed        uint64      `json:"elapsed"`
 	Count          int         `json:"count"`
 	TimeRangeCount int         `json:"timeRangeCount"`
 	TotalCount     int         `json:"totalCount"`
+	timestamp      uint64
+	elapsed        uint64
 }
 
 type QueryParams struct {
@@ -430,6 +430,7 @@ type QueryParams struct {
 	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"`
@@ -442,49 +443,60 @@ func getQueryParams(url *url.URL) QueryParams {
 		Limit:              -1,
 		SortBy:             "identity.asc",
 		Filter:             "",
+		FilterFields:       make(map[string]string),
 		TimeRangeStart:     now - (15 * oneMinute),
 		TimeRangeEnd:       now,
 		TimeRangeOperation: intersects,
 	}
-	offset, err := strconv.Atoi(url.Query().Get("offset"))
-	if err == nil {
-		qp.Offset = offset
-	}
-	limit, err := strconv.Atoi(url.Query().Get("limit"))
-	if err == nil {
-		qp.Limit = limit
-	}
-	sortBy := url.Query().Get("sortBy")
-	if sortBy != "" {
-		qp.SortBy = sortBy
-	}
-	filter := url.Query().Get("filter")
-	if filter != "" {
-		qp.Filter = filter
-	}
-	timeRangeStart := url.Query().Get("timeRangeStart")
-	if timeRangeStart != "" {
-		v, err := strconv.Atoi(timeRangeStart)
-		if err == nil {
-			qp.TimeRangeStart = uint64(v)
-		}
-	}
-	timeRangeEnd := url.Query().Get("timeRangeEnd")
-	if timeRangeEnd != "" {
-		v, err := strconv.Atoi(timeRangeEnd)
-		if err == nil {
-			qp.TimeRangeEnd = uint64(v)
+
+	for k, v := range url.Query() {
+		switch k {
+		case "offset":
+			offset, err := strconv.Atoi(v[0])
+			if err == nil {
+				qp.Offset = offset
+			}
+		case "limit":
+			limit, err := strconv.Atoi(v[0])
+			if err == nil {
+				qp.Limit = limit
+			}
+		case "sortBy":
+			if v[0] != "" {
+				qp.SortBy = v[0]
+			}
+		case "filter":
+			if v[0] != "" {
+				qp.Filter = v[0]
+			}
+		case "timeRangeStart":
+			if v[0] != "" {
+				v, err := strconv.Atoi(v[0])
+				if err == nil {
+					qp.TimeRangeStart = uint64(v)
+				}
+			}
+		case "timeRangeEnd":
+			if v[0] != "" {
+				v, err := strconv.Atoi(v[0])
+				if err == nil {
+					qp.TimeRangeEnd = uint64(v)
+				}
+			}
+		case "timeRangeOperation":
+			timeRangeOperation := v[0]
+			switch timeRangeOperation {
+			case "contains":
+				qp.TimeRangeOperation = contains
+			case "within":
+				qp.TimeRangeOperation = within
+			default:
+				qp.TimeRangeOperation = intersects
+			}
+		default:
+			qp.FilterFields[cases.Title(language.Und, cases.NoLower).String(k)] = v[0]
 		}
 	}
-	timeRangeOperation := url.Query().Get("timeRangeOperation")
-	switch timeRangeOperation {
-	case "contains":
-		qp.TimeRangeOperation = contains
-	case "within":
-		qp.TimeRangeOperation = within
-	default:
-		qp.TimeRangeOperation = intersects
-	}
 	return qp
 }
 
@@ -569,6 +581,19 @@ func validateAndReturnFilterQuery(filter string) (string, string, error) {
 	return field, match, nil
 }
 
+func validateAndReturnFilterFieldQuery(filterField string) (string, string, error) {
+	parts := strings.Split(filterField, ".")
+	if len(parts) > 2 {
+		return "", "", fmt.Errorf("Malformed filter field query value parameter")
+	}
+	field := cases.Title(language.Und, cases.NoLower).String(parts[0])
+	subField := ""
+	if len(parts) == 2 {
+		subField = cases.Title(language.Und, cases.NoLower).String(parts[1])
+	}
+	return field, subField, nil
+}
+
 func getField(field string, record interface{}) interface{} {
 	x := reflect.ValueOf(record)
 	if x.Kind() == reflect.Struct {
@@ -672,7 +697,53 @@ func compareFields(x, y interface{}, order string) bool {
 }
 
 // type any = interface{}
-func filterRecord[T any](item T, filter string) bool {
+func filterRecord[T any](item T, qp QueryParams) bool {
+	filter := true
+	if qp.Filter != "" {
+		field, match, err := validateAndReturnFilterQuery(qp.Filter)
+		// todo propagate error or log
+		if err != nil {
+			return false
+		}
+		value := getField(field, item)
+		if value == nil {
+			filter = false
+		} else {
+			x := reflect.ValueOf(value)
+			if x.Kind() == reflect.Struct {
+				if !filterSubRecord(value, match) {
+					filter = false
+				}
+			} else if !matchFieldValue(value, match) {
+				filter = false
+			}
+		}
+	} else if len(qp.FilterFields) > 0 {
+		for field, match := range qp.FilterFields {
+			field, subField, err := validateAndReturnFilterFieldQuery(field)
+			if err != nil {
+				return false
+			}
+			value := getField(field, item)
+			if value == nil {
+				filter = false
+			} else {
+				x := reflect.ValueOf(value)
+				if x.Kind() == reflect.Struct {
+					if !filterSubRecord(value, subField+"."+match) {
+						filter = false
+					}
+				} else if !matchFieldValue(value, match) {
+					filter = false
+				}
+			}
+		}
+	}
+	return filter
+}
+
+// type any = interface{}
+func filterSubRecord[T any](item T, filter string) bool {
 	if filter == "" {
 		return true
 	}
@@ -684,7 +755,7 @@ func filterRecord[T any](item T, filter string) bool {
 	value := getField(field, item)
 	x := reflect.ValueOf(value)
 	if x.Kind() == reflect.Struct {
-		return filterRecord(value, match)
+		return filterSubRecord(value, match)
 	}
 	return matchFieldValue(value, match)
 }


Test patch:
diff --git a/client/podman/informer_test.go b/client/podman/informer_test.go
new file mode 100644
index 00000000..08e57b90
--- /dev/null
+++ b/client/podman/informer_test.go
@@ -0,0 +1,131 @@
+package podman
+
+import (
+	"context"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/skupperproject/skupper/client/container"
+	"github.com/skupperproject/skupper/pkg/images"
+	"github.com/skupperproject/skupper/pkg/utils"
+	"gotest.tools/assert"
+)
+
+func TestContainerInformer(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cli, wg := NewClientOrSkip(t, ctx)
+	stopCh := make(chan struct{})
+	defer wg.Wait()
+	defer cancel()
+	defer close(stopCh)
+
+	ci := NewContainerInformer(cli)
+	ci.SetResyncPeriod(time.Second)
+	ci.Start(stopCh)
+
+	namePrefix := RandomName("container")
+
+	type result struct {
+		created []string
+		updated []string
+		deleted []string
+	}
+
+	scenarios := []struct {
+		name   string
+		create []string
+		update []string
+		delete []string
+	}{
+		{
+			name:   "add",
+			create: []string{namePrefix + "1", namePrefix + "2", namePrefix + "3"},
+		},
+		{
+			name:   "add-remove",
+			create: []string{namePrefix + "4"},
+			delete: []string{namePrefix + "1"},
+		},
+		{
+			name:   "update-remove",
+			update: []string{namePrefix + "2"},
+			delete: []string{namePrefix + "3"},
+		},
+		{
+			name:   "add-update-remove",
+			create: []string{namePrefix + "1", namePrefix + "3"},
+			update: []string{namePrefix + "2", namePrefix + "4"},
+			delete: []string{namePrefix + "4"},
+		},
+		{
+			name:   "remove",
+			delete: []string{namePrefix + "1", namePrefix + "2", namePrefix + "3"},
+		},
+	}
+
+	// Pulling image
+	image := images.GetServiceControllerImageName()
+	assert.Assert(t, cli.ImagePull(image))
+
+	var containersLeft = map[string]bool{}
+	for _, s := range scenarios {
+		t.Run(s.name, func(t *testing.T) {
+			var res result
+			ci.AddInformer(&container.InformerBase[*container.Container]{
+				Add: func(obj *container.Container) {
+					if strings.HasPrefix(obj.Name, namePrefix) {
+						res.created = append(res.created, obj.Name)
+					}
+				},
+				Update: func(oldObj, newObj *container.Container) {
+					if strings.HasPrefix(newObj.Name, namePrefix) {
+						res.updated = append(res.updated, newObj.Name)
+					}
+				},
+				Delete: func(obj *container.Container) {
+					if strings.HasPrefix(obj.Name, namePrefix) {
+						res.deleted = append(res.deleted, obj.Name)
+					}
+				},
+			})
+
+			// Creating containers
+			for _, c := range s.create {
+				assert.Assert(t, cli.ContainerCreate(&container.Container{
+					Name:    c,
+					Image:   image,
+					Command: []string{"tail", "-f", "/dev/null"},
+				}))
+				containersLeft[c] = true
+				assert.Assert(t, cli.ContainerStart(c))
+			}
+
+			// Restarting containers (update)
+			for _, u := range s.update {
+				assert.Assert(t, cli.ContainerRestart(u))
+			}
+
+			// Deleting containers
+			for _, d := range s.delete {
+				assert.Assert(t, cli.ContainerRemove(d))
+				delete(containersLeft, d)
+			}
+
+			// Waiting 30 secs for state to match
+			assert.Assert(t, utils.Retry(time.Second, 30, func() (bool, error) {
+				match := reflect.DeepEqual(s.create, res.created) &&
+					reflect.DeepEqual(s.update, res.updated) &&
+					reflect.DeepEqual(s.delete, res.deleted)
+				return match, nil
+			}), "Created { expected: %v, got: %v } - Updated { expected: %v, got: %v } - Deleted { expected: %v, got: %v }",
+				s.create, res.created, s.update, res.updated, s.delete, res.deleted)
+		})
+	}
+
+	for c, _ := range containersLeft {
+		t.Logf("removing container left: %s", c)
+		_ = cli.ContainerRemove(c)
+	}
+}
diff --git a/pkg/flow/record_test.go b/pkg/flow/record_test.go
index 27c701ce..70273074 100644
--- a/pkg/flow/record_test.go
+++ b/pkg/flow/record_test.go
@@ -554,7 +554,62 @@ func TestFilterRecord(t *testing.T) {
 	}
 
 	for _, test := range testTable {
-		result := filterRecord(flowPair, test.filter)
+		qp := QueryParams{Filter: test.filter}
+		result := filterRecord(flowPair, qp)
+		assert.Equal(t, result, test.result)
+	}
+}
+
+func TestFilterFieldsRecord(t *testing.T) {
+	host := "10.20.30.40"
+	name := "public1"
+	octets := uint64(2030)
+	flow := FlowRecord{
+		SourceHost: &host,
+		Octets:     &octets,
+	}
+	flowPair := FlowPairRecord{
+		Base: Base{
+			Identity: "foo",
+		},
+		SourceSiteName: &name,
+		ForwardFlow:    &flow,
+	}
+
+	type test struct {
+		filterField map[string]string
+		result      bool
+	}
+	testTable := []test{
+		{
+			filterField: map[string]string{"forwardFlow.SourceHost": "10.20.30.40"},
+			result:      true,
+		},
+		{
+			filterField: map[string]string{"sourceSiteName": "public1"},
+			result:      true,
+		},
+		{
+			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"},
+			result:      false,
+		},
+		{
+			filterField: map[string]string{},
+			result:      true,
+		},
+		{
+			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)
 	}
 }