Backport wathola configurability (#921)

* Flags to configure eventing upgrade tests forwarder (ksvc) (knative#3899)

* Flags to configure eventing upgrade tests forwarder (ksvc)

This change is required to complete knative/operator#252

It enables configuring of wathola forwarder implemented as Knative
Service. Configuration is done by using new optional environment
variables.

* Using kelseyhightower/envconfig after code review

* Docs for Eventing upgrade tests config overrides.

* Eventing upgrade tests prober fully configurable

Conflicts resolved:
 * test/upgrade/prober/configuration.go
 * test/upgrade/prober/forwarder.go

test/upgrade/README.md

@@ -83,3 +83,23 @@ Diagram below describe the setup:
           +--------+       +---------+ |
            (default)        +----------+
 ```
+
+#### Probe test configuration
+
+Probe test behavior can be influenced from outside without modifying its source
+code. That can be beneficial if one would like to run upgrade tests in different
+context. One such example might be running Eventing upgrade tests in place that
+have Serving and Eventing both installed. In such environment one can set
+environment variable `E2E_UPGRADE_TESTS_SERVING_USE` to enable usage of ksvc
+forwarder (which is disabled by default):
+
+```
+$ export E2E_UPGRADE_TESTS_SERVING_USE=true
+```
+
+Any option, apart from namespace, in
+[`knative.dev/eventing/test/upgrade/prober.Config`](https://github.com/knative/eventing/blob/022e281/test/upgrade/prober/prober.go#L52-L63)
+struct can be influenced, by using `E2E_UPGRADE_TESTS_XXXXX` environmental
+variable prefix (using
+[kelseyhightower/envconfig](https://github.com/kelseyhightower/envconfig#usage)
+usage).

test/upgrade/probe_test.go

@@ -51,9 +51,6 @@ func TestContinuousEventsPropagationWithProber(t *testing.T) {
 
 	config := prober.NewConfig(client.Namespace)
 
-	// FIXME: https://github.com/knative/eventing/issues/2665
-	config.FailOnErrors = false
-
 	// Use zap.SugarLogger instead of t.Logf because we want to see failures
 	// inline with other logs instead of buffered until the end.
 	log := createLogger()

test/upgrade/prober/config.toml

@@ -1,6 +1,6 @@
 # logLevel = 5 # DEBUG(5)
 [sender]
-address = '{{- .BrokerUrl -}}'
+address = '{{- .BrokerURL -}}'
 interval = {{ .Config.Interval.Nanoseconds }}
 [forwarder]
 target = 'http://wathola-receiver.{{- .Config.Namespace -}}.svc.cluster.local'

test/upgrade/prober/configuration.go

@@ -22,7 +22,9 @@ import (
 	"path"
 	"runtime"
 	"text/template"
+	"time"
 
+	"github.com/kelseyhightower/envconfig"
 	"github.com/wavesoftware/go-ensure"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	eventingv1beta1 "knative.dev/eventing/pkg/apis/eventing/v1beta1"
@@ -33,17 +35,82 @@ import (
 )
 
 const (
-	configName       = "wathola-config"
-	configMountPoint = "/home/nonroot/.config/wathola"
-	configFilename   = "config.toml"
-	watholaEventNs   = "com.github.cardil.wathola"
-	healthEndpoint   = "/healthz"
+	defaultConfigName          = "wathola-config"
+	defaultConfigHomedirPath   = ".config/wathola"
+	defaultHomedir             = "/home/nonroot"
+	defaultConfigFilename      = "config.toml"
+	defaultWatholaEventsPrefix = "com.github.cardil.wathola"
+	defaultBrokerName          = "default"
+	defaultHealthEndpoint      = "/healthz"
+	defaultFinishedSleep       = 5 * time.Second
 )
 
-var (
-	eventTypes = []string{"step", "finished"}
-	brokerName = "default"
-)
+var eventTypes = []string{"step", "finished"}
+
+// Config represents a configuration for prober.
+type Config struct {
+	Wathola       WatholaConfig
+	Namespace     string
+	Interval      time.Duration
+	FinishedSleep time.Duration
+	Serving       ServingConfig
+	FailOnErrors  bool
+}
+
+// WatholaConfig represents options related strictly to wathola testing tool.
+type WatholaConfig struct {
+	Config           ConfigMapConfig
+	EventsTypePrefix string
+	HealthEndpoint   string
+	BrokerName       string
+}
+
+// ConfigMapConfig represents options of wathola config toml file.
+type ConfigMapConfig struct {
+	Name       string
+	MountPoint string
+	Filename   string
+}
+
+// ServingConfig represents a options for serving test component (wathola-forwarder).
+type ServingConfig struct {
+	Use         bool
+	ScaleToZero bool
+}
+
+// NewConfig creates a new configuration object with default values filled in.
+// Values can be influenced by kelseyhightower/envconfig with
+// `e2e_upgrade_tests` prefix.
+func NewConfig(namespace string) *Config {
+	config := &Config{
+		Namespace:     "",
+		Interval:      Interval,
+		FinishedSleep: defaultFinishedSleep,
+		FailOnErrors:  true,
+		Serving: ServingConfig{
+			Use:         false,
+			ScaleToZero: true,
+		},
+		Wathola: WatholaConfig{
+			Config: ConfigMapConfig{
+				Name:       defaultConfigName,
+				MountPoint: fmt.Sprintf("%s/%s", defaultHomedir, defaultConfigHomedirPath),
+				Filename:   defaultConfigFilename,
+			},
+			EventsTypePrefix: defaultWatholaEventsPrefix,
+			HealthEndpoint:   defaultHealthEndpoint,
+			BrokerName:       defaultBrokerName,
+		},
+	}
+
+	// FIXME: remove while fixing https://github.com/knative/eventing/issues/2665
+	config.FailOnErrors = false
+
+	err := envconfig.Process("e2e_upgrade_tests", config)
+	ensure.NoError(err)
+	config.Namespace = namespace
+	return config
+}
 
 func (p *prober) deployConfiguration() {
 	p.deployBroker()
@@ -52,55 +119,64 @@ func (p *prober) deployConfiguration() {
 }
 
 func (p *prober) deployBroker() {
-	p.client.CreateBrokerV1Beta1OrFail(brokerName)
+	p.client.CreateBrokerV1Beta1OrFail(p.config.Wathola.BrokerName)
 }
 
-func (p *prober) fetchBrokerUrl() (*apis.URL, error) {
+func (p *prober) fetchBrokerURL() (*apis.URL, error) {
 	namespace := p.config.Namespace
-	p.log.Debugf("Fetching %s broker URL for ns %s", brokerName, namespace)
-	meta := resources.NewMetaResource(brokerName, p.config.Namespace, testlib.BrokerTypeMeta)
+	p.log.Debugf("Fetching %s broker URL for ns %s",
+		p.config.Wathola.BrokerName, namespace)
+	meta := resources.NewMetaResource(
+		p.config.Wathola.BrokerName, p.config.Namespace, testlib.BrokerTypeMeta,
+	)
 	err := duck.WaitForResourceReady(p.client.Dynamic, meta)
 	if err != nil {
 		return nil, err
 	}
-	broker, err := p.client.Eventing.EventingV1beta1().Brokers(namespace).Get(brokerName, metav1.GetOptions{})
+	broker, err := p.client.Eventing.EventingV1beta1().Brokers(namespace).Get(
+		p.config.Wathola.BrokerName, metav1.GetOptions{},
+	)
 	if err != nil {
 		return nil, err
 	}
 	url := broker.Status.Address.URL
-	p.log.Debugf("%s broker URL for ns %s is %v", brokerName, namespace, url)
+	p.log.Debugf("%s broker URL for ns %s is %v",
+		p.config.Wathola.BrokerName, namespace, url)
 	return url, nil
 }
 
 func (p *prober) deployConfigMap() {
-	name := configName
+	name := p.config.Wathola.Config.Name
 	p.log.Infof("Deploying config map: \"%s/%s\"", p.config.Namespace, name)
-	brokerUrl, err := p.fetchBrokerUrl()
+	brokerURL, err := p.fetchBrokerURL()
 	ensure.NoError(err)
-	configData := p.compileTemplate(configFilename, brokerUrl)
-	p.client.CreateConfigMapOrFail(name, p.config.Namespace, map[string]string{configFilename: configData})
+	configData := p.compileTemplate(p.config.Wathola.Config.Filename, brokerURL)
+	p.client.CreateConfigMapOrFail(name, p.config.Namespace, map[string]string{
+		p.config.Wathola.Config.Filename: configData,
+	})
 }
 
 func (p *prober) deployTriggers() {
 	for _, eventType := range eventTypes {
 		name := fmt.Sprintf("wathola-trigger-%v", eventType)
-		fullType := fmt.Sprintf("%v.%v", watholaEventNs, eventType)
+		fullType := fmt.Sprintf("%v.%v", p.config.Wathola.EventsTypePrefix, eventType)
 		subscriberOption := resources.WithSubscriberServiceRefForTriggerV1Beta1(receiverName)
 		if p.config.Serving.Use {
 			subscriberOption = resources.WithSubscriberKServiceRefForTrigger(forwarderName)
 		}
 		p.client.CreateTriggerOrFailV1Beta1(name,
-			resources.WithBrokerV1Beta1(brokerName),
+			resources.WithBrokerV1Beta1(p.config.Wathola.BrokerName),
 			resources.WithAttributesTriggerFilterV1Beta1(
 				eventingv1beta1.TriggerAnyFilter,
 				fullType,
 				map[string]interface{}{},
 			),
-			subscriberOption)
+			subscriberOption,
+		)
 	}
 }
 
-func (p *prober) compileTemplate(templateName string, brokerUrl *apis.URL) string {
+func (p *prober) compileTemplate(templateName string, brokerURL fmt.Stringer) string {
 	_, filename, _, _ := runtime.Caller(0)
 	templateFilepath := path.Join(path.Dir(filename), templateName)
 	templateBytes, err := ioutil.ReadFile(templateFilepath)
@@ -110,10 +186,10 @@ func (p *prober) compileTemplate(templateName string, brokerUrl *apis.URL) strin
 	var buff bytes.Buffer
 	data := struct {
 		*Config
-		BrokerUrl string
+		BrokerURL string
 	}{
 		p.config,
-		brokerUrl.String(),
+		brokerURL.String(),
 	}
 	ensure.NoError(tmpl.Execute(&buff, data))
 	return buff.String()

test/upgrade/prober/forwarder.go

@@ -34,7 +34,7 @@ var (
 func (p *prober) deployForwarder() {
 	p.log.Infof("Deploy forwarder knative service: %v", forwarderName)
 	serving := p.client.Dynamic.Resource(resources.KServicesGVR).Namespace(p.client.Namespace)
-	service := forwarderKService(forwarderName, p.client.Namespace)
+	service := p.forwarderKService(forwarderName, p.client.Namespace)
 	_, err := serving.Create(service, metav1.CreateOptions{})
 	ensure.NoError(err)
 
@@ -59,7 +59,7 @@ func (p *prober) removeForwarder() {
 	ensure.NoError(err)
 }
 
-func forwarderKService(name, namespace string) *unstructured.Unstructured {
+func (p *prober) forwarderKService(name, namespace string) *unstructured.Unstructured {
 	obj := map[string]interface{}{
 		"apiVersion": resources.KServiceType.APIVersion,
 		"kind":       resources.KServiceType.Kind,
@@ -77,20 +77,20 @@ func forwarderKService(name, namespace string) *unstructured.Unstructured {
 						"name":  "forwarder",
 						"image": pkgTest.ImagePath(forwarderName),
 						"volumeMounts": []map[string]interface{}{{
-							"name":      configName,
-							"mountPath": configMountPoint,
+							"name":      p.config.Wathola.Config.Name,
+							"mountPath": p.config.Wathola.Config.MountPoint,
 							"readOnly":  true,
 						}},
 						"readinessProbe": map[string]interface{}{
 							"httpGet": map[string]interface{}{
-								"path": healthEndpoint,
+								"path": p.config.Wathola.HealthEndpoint,
 							},
 						},
 					}},
 					"volumes": []map[string]interface{}{{
-						"name": configName,
+						"name": p.config.Wathola.Config.Name,
 						"configMap": map[string]interface{}{
-							"name": configName,
+							"name": p.config.Wathola.Config.Name,
 						},
 					}},
 				},

test/upgrade/prober/prober.go

@@ -48,33 +48,6 @@ type Prober interface {
 	remove()
 }
 
-// Config represents a configuration for prober
-type Config struct {
-	Namespace     string
-	Interval      time.Duration
-	Serving       ServingConfig
-	FinishedSleep time.Duration
-	FailOnErrors  bool
-}
-
-type ServingConfig struct {
-	Use         bool
-	ScaleToZero bool
-}
-
-func NewConfig(namespace string) *Config {
-	return &Config{
-		Namespace:     namespace,
-		Interval:      Interval,
-		FinishedSleep: 5 * time.Second,
-		FailOnErrors:  true,
-		Serving: ServingConfig{
-			Use:         false,
-			ScaleToZero: true,
-		},
-	}
-}
-
 // RunEventProber starts a single Prober of the given domain.
 func RunEventProber(log *zap.SugaredLogger, client *testlib.Client, config *Config) Prober {
 	pm := newProber(log, client, config)
@@ -151,7 +124,7 @@ func (p *prober) remove() {
 	if p.config.Serving.Use {
 		p.removeForwarder()
 	}
-	p.client.Tracker.Clean(true)
+	ensure.NoError(p.client.Tracker.Clean(true))
 }
 
 func newProber(log *zap.SugaredLogger, client *testlib.Client, config *Config) Prober {