add spec.observer (auto|none) and wire to router (#2348)

* add spec.observer (auto|none) and wire to router

Fixes #2342

* Added unit tests

* Added http1 and http2 values to spec.observer; updated unit tests

* Update config/crd/bases/skupper_listener_crd.yaml

Co-authored-by: Paul Wright <5154224+pwright@users.noreply.github.com>

---------

Co-authored-by: Paul Wright <5154224+pwright@users.noreply.github.com>

Author: AryanP123

config/crd/bases/skupper_listener_crd.yaml

@@ -22,6 +22,18 @@ spec:
             spec:
               type: object
               properties:
+                observer:
+                  description: |-
+                     The listener observer controls how the listener inspects network traffic for  application-level protocol information.
+                      When unset or set to `auto`, the listener inspects traffic to detect known  application protocols and produces telemetry events for that application  traffic.
+                     Set to a specific protocol (`http1` or `http2`) to restrict inspection to that  protocol only.
+                     Set to `none` to disable protocol inspection and reduce overhead from traffic  inspection and application-level telemetry.
+                  type: string
+                  enum:
+                    - auto
+                    - none
+                    - http1
+                    - http2
                 routingKey:
                   description: |-
                     The identifier to route traffic from listeners to connectors. To

internal/qdr/amqp_mgmt.go

@@ -130,6 +130,7 @@ func asTcpEndpoint(record Record) TcpEndpoint {
 		Address:    record.AsString("address"),
 		SiteId:     record.AsString("siteId"),
 		SslProfile: record.AsString("sslProfile"),
+		Observer:   record.AsString("observer"),
 		ProcessID:  record.AsString("processId"),
 	}
 	if value, ok := record["verifyHostname"]; ok {

internal/qdr/qdr.go

@@ -559,6 +559,7 @@ type TcpEndpoint struct {
 	Address        string `json:"address,omitempty"`
 	SiteId         string `json:"siteId,omitempty"`
 	SslProfile     string `json:"sslProfile,omitempty"`
+	Observer       string `json:"observer,omitempty"`
 	VerifyHostname *bool  `json:"verifyHostname,omitempty"`
 	ProcessID      string `json:"processId,omitempty"`
 }
@@ -583,6 +584,9 @@ func (e TcpEndpoint) toRecord() Record {
 	if e.SslProfile != "" {
 		result["sslProfile"] = e.SslProfile
 	}
+	if e.Observer != "" {
+		result["observer"] = e.Observer
+	}
 	if e.VerifyHostname != nil {
 		result["verifyHostname"] = e.VerifyHostname
 	}
@@ -948,8 +952,17 @@ func (a TcpEndpoint) equivalentVerifyHostname(b TcpEndpoint) bool {
 }
 
 func (a TcpEndpoint) Equivalent(b TcpEndpoint) bool {
+	obsA := a.Observer
+	if obsA == "" {
+		obsA = "auto"
+	}
+	obsB := b.Observer
+	if obsB == "" {
+		obsB = "auto"
+	}
 	if !equivalentHost(a.Host, b.Host) || a.Port != b.Port || a.Address != b.Address ||
-		a.SiteId != b.SiteId || a.ProcessID != b.ProcessID || !a.equivalentVerifyHostname(b) {
+		a.SiteId != b.SiteId || a.ProcessID != b.ProcessID || !a.equivalentVerifyHostname(b) ||
+		obsA != obsB {
 		return false
 	}
 	return true

internal/qdr/qdr_test.go

@@ -668,3 +668,91 @@ func TestRecordTypes_GH2081(t *testing.T) {
 		})
 	}
 }
+
+func TestTcpEndpointObserverToRecord(t *testing.T) {
+	// empty observer -> omitted
+	e := TcpEndpoint{
+		Name:     "t1",
+		Address:  "a",
+		Host:     "h",
+		Port:     "1234",
+		SiteId:   "s",
+		Observer: "",
+	}
+	r := e.toRecord()
+	_, ok := r["observer"]
+	assert.Assert(t, !ok, "observer should be omitted when empty")
+
+	// non-empty observer -> included
+	for _, val := range []string{"auto", "none", "http1", "http2"} {
+		e.Observer = val
+		r = e.toRecord()
+		got, ok := r["observer"]
+		assert.Assert(t, ok, "observer should be present when set")
+		assert.Equal(t, got, val)
+	}
+}
+
+func TestTcpEndpointEquivalentObserverAutoVsEmpty(t *testing.T) {
+	a := TcpEndpoint{
+		Name:     "t1",
+		Address:  "a",
+		Host:     "",
+		Port:     "1234",
+		SiteId:   "s",
+		Observer: "auto",
+	}
+	b := TcpEndpoint{
+		Name:     "t1",
+		Address:  "a",
+		Host:     "",
+		Port:     "1234",
+		SiteId:   "s",
+		Observer: "",
+	}
+	if !a.Equivalent(b) {
+		t.Errorf("expected endpoints to be equivalent when comparing observer 'auto' vs empty")
+	}
+}
+
+func TestTcpEndpointEquivalentHttpModes(t *testing.T) {
+	base := TcpEndpoint{
+		Name:    "t1",
+		Address: "a",
+		Host:    "",
+		Port:    "1234",
+		SiteId:  "s",
+	}
+	// Equal http1/http1
+	a := base
+	b := base
+	a.Observer = "http1"
+	b.Observer = "http1"
+	if !a.Equivalent(b) {
+		t.Errorf("expected http1 vs http1 to be equivalent")
+	}
+	// Equal http2/http2
+	a.Observer = "http2"
+	b.Observer = "http2"
+	if !a.Equivalent(b) {
+		t.Errorf("expected http2 vs http2 to be equivalent")
+	}
+	// Different http1 vs auto
+	a.Observer = "http1"
+	b.Observer = "auto"
+	if a.Equivalent(b) {
+		t.Errorf("expected http1 vs auto to be not equivalent")
+	}
+	// Different http2 vs empty (router default)
+	a.Observer = "http2"
+	b.Observer = ""
+	if a.Equivalent(b) {
+		t.Errorf("expected http2 vs empty to be not equivalent")
+	}
+	// Different http1 vs none
+	a.Observer = "http1"
+	b.Observer = "none"
+	if a.Equivalent(b) {
+		t.Errorf("expected http1 vs none to be not equivalent")
+	}
+}

internal/site/listener.go

@@ -21,6 +21,7 @@ func UpdateBridgeConfigForListenerWithHostAndPort(siteId string, listener *skupp
 			Port:       strconv.Itoa(port),
 			Address:    listener.Spec.RoutingKey,
 			SslProfile: listener.Spec.TlsCredentials,
+			Observer:   listener.Spec.Observer,
 		})
 	}
 }

pkg/apis/skupper/v2alpha1/types.go

@@ -407,6 +407,7 @@ type ListenerSpec struct {
 	Port             int               `json:"port"`
 	TlsCredentials   string            `json:"tlsCredentials,omitempty"`
 	Type             string            `json:"type,omitempty"`
+	Observer         string            `json:"observer,omitempty"`
 	ExposePodsByName bool              `json:"exposePodsByName,omitempty"`
 	Settings         map[string]string `json:"settings,omitempty"`
 }