Repository: knative/pkg
Pull request #1439 (2020-06-23 13:42:18)

use coordination.Leases exclusively for leader election
Prior locks were for K8s version <1.14

Fixes: #1427

I left the plumbing of the lock setting through everything since there may be a new lock in the future we would want to switch to - if folks feel like this is unnecessary let me know and I can rip it out

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1439#" title="Author self-approved">dprotaso</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **evankanderson**
You can assign the PR to them by writing `/assign @evankanderson` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative/pkg/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["evankanderson"]} -->
dprotaso: /assign @pmorie @mattmoor 
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1439#" title="Author self-approved">dprotaso</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **mattmoor**
You can assign the PR to them by writing `/assign @mattmoor` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative/pkg/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["mattmoor"]} -->
markusthoemmes: Is this a "breaking" change? Do we need to check if anybody is using something different than leases first?
knative-prow-robot: @dprotaso: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-pkg-unit-tests | 274a32ba7c697b4d6e52e147d2ca29956d4c5acd | [link](https://prow.knative.dev/view/gcs/knative-prow/pr-logs/pull/knative_pkg/1439/pull-knative-pkg-unit-tests/1277610416846934016) | `/test pull-knative-pkg-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative_pkg/1439). [Your PR dashboard](https://gubernator.knative.dev/pr/dprotaso).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
dprotaso: > Is this a "breaking" change? Do we need to check if anybody is using something different than leases first?

Technically this feature is alpha so breaking changes are allowed
mattmoor: According to @pmorie (in slack) the Operator doesn't reference this field, and the docs I just updated don't mention configuring this.  I believe we've also established that all supported K8s versions have had the coordination API (probably since the inception of this feature).

I agree that this is breaking, but also that it's alpha and now's the time to make breaking changes.  @markusthoemmes beyond this simply being available in prior releases, do you have reason to think this might present problems?

Patch:
diff --git a/leaderelection/config.go b/leaderelection/config.go
index 8e0728e1f..880dd8c2b 100644
--- a/leaderelection/config.go
+++ b/leaderelection/config.go
@@ -27,6 +27,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 
 	cm "knative.dev/pkg/configmap"
 )
@@ -37,7 +38,7 @@ const configMapNameEnv = "CONFIG_LEADERELECTION_NAME"
 // This is a variable so that it may be customized in the binary entrypoint.
 var MaxBuckets uint32 = 10
 
-var validResourceLocks = sets.NewString("leases", "configmaps", "endpoints")
+var validResourceLocks = sets.NewString(resourcelock.LeasesResourceLock)
 
 // NewConfigFromMap returns a Config for the given map, or an error.
 func NewConfigFromMap(data map[string]string) (*Config, error) {
@@ -64,7 +65,7 @@ func NewConfigFromMap(data map[string]string) (*Config, error) {
 		return nil, fmt.Errorf("buckets: value must be between %d <= %d <= %d", 1, config.Buckets, MaxBuckets)
 	}
 	if !validResourceLocks.Has(config.ResourceLock) {
-		return nil, fmt.Errorf(`resourceLock: invalid value %q: valid values are "leases","configmaps","endpoints"`, config.ResourceLock)
+		return nil, fmt.Errorf(`resourceLock: invalid value %q: valid values are "leases"`, config.ResourceLock)
 	}
 
 	return config, nil


Test patch:
diff --git a/leaderelection/config_test.go b/leaderelection/config_test.go
index e2850c8a8..c9066ca4c 100644
--- a/leaderelection/config_test.go
+++ b/leaderelection/config_test.go
@@ -92,7 +92,7 @@ func TestNewConfigMapFromData(t *testing.T) {
 		data: kmeta.UnionMaps(okData(), map[string]string{
 			"resourceLock": "flarps",
 		}),
-		err: errors.New(`resourceLock: invalid value "flarps": valid values are "leases","configmaps","endpoints"`),
+		err: errors.New(`resourceLock: invalid value "flarps": valid values are "leases"`),
 	}, {
 		name: "invalid leaseDuration",
 		data: kmeta.UnionMaps(okData(), map[string]string{
diff --git a/test/zipkin/util.go b/test/zipkin/util.go
index 1ff2050a7..b5b574911 100644
--- a/test/zipkin/util.go
+++ b/test/zipkin/util.go
@@ -23,11 +23,18 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
 	"sync"
+	"testing"
 	"time"
 
+	tracingconfig "knative.dev/pkg/tracing/config"
+
 	"github.com/openzipkin/zipkin-go/model"
 	"go.opencensus.io/trace"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"knative.dev/pkg/test/logging"
 	"knative.dev/pkg/test/monitoring"
@@ -46,10 +53,7 @@ const (
 
 	// App is the name of this component.
 	// This will be used as a label selector.
-	app = "zipkin"
-
-	// istioNS is the namespace we are using for istio components.
-	istioNS = "istio-system"
+	appLabel = "zipkin"
 )
 
 var (
@@ -65,26 +69,66 @@ var (
 	teardownOnce sync.Once
 )
 
+// SetupZipkinTracingFromConfigTracing setups zipkin tracing like SetupZipkinTracing but retrieving the zipkin configuration
+// from config-tracing config map
+func SetupZipkinTracingFromConfigTracing(kubeClientset *kubernetes.Clientset, logf logging.FormatLogger, configMapNamespace string) error {
+	cm, err := kubeClientset.CoreV1().ConfigMaps(configMapNamespace).Get("config-tracing", metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("error while retrieving config-tracing config map: %w", err)
+	}
+	c, err := tracingconfig.NewTracingConfigFromConfigMap(cm)
+	if err != nil {
+		return fmt.Errorf("error while parsing config-tracing config map: %w", err)
+	}
+	zipkinEndpointURL, err := url.Parse(c.ZipkinEndpoint)
+	if err != nil {
+		return fmt.Errorf("error while parsing the zipkin endpoint in config-tracing config map: %w", err)
+	}
+	unparsedPort := zipkinEndpointURL.Port()
+	port := uint64(80)
+	if unparsedPort != "" {
+		port, err = strconv.ParseUint(unparsedPort, 10, 16)
+		if err != nil {
+			return fmt.Errorf("error while parsing the zipkin endpoint port in config-tracing config map: %w", err)
+		}
+	}
+
+	namespace, err := parseNamespaceFromHostname(zipkinEndpointURL.Host)
+	if err != nil {
+		return fmt.Errorf("error while parsing the Zipkin endpoint in config-tracing config map: %w", err)
+	}
+
+	return SetupZipkinTracing(kubeClientset, logf, int(port), namespace)
+}
+
+// SetupZipkinTracingFromConfigTracingOrFail is same as SetupZipkinTracingFromConfigTracing, but fails the test if an error happens
+func SetupZipkinTracingFromConfigTracingOrFail(t testing.TB, kubeClientset *kubernetes.Clientset, configMapNamespace string) {
+	if err := SetupZipkinTracingFromConfigTracing(kubeClientset, t.Logf, configMapNamespace); err != nil {
+		t.Fatalf("Error while setup Zipkin tracing: %v", err)
+	}
+}
+
 // SetupZipkinTracing sets up zipkin tracing which involves:
 // 1. Setting up port-forwarding from localhost to zipkin pod on the cluster
 //    (pid of the process doing Port-Forward is stored in a global variable).
 // 2. Enable AlwaysSample config for tracing for the SpoofingClient.
-func SetupZipkinTracing(kubeClientset *kubernetes.Clientset, logf logging.FormatLogger) bool {
+// The zipkin deployment must have the label app=zipkin
+func SetupZipkinTracing(kubeClientset *kubernetes.Clientset, logf logging.FormatLogger, zipkinRemotePort int, zipkinNamespace string) (err error) {
 	setupOnce.Do(func() {
-		if err := monitoring.CheckPortAvailability(ZipkinPort); err != nil {
-			logf("Zipkin port not available on the machine: %v", err)
+		if e := monitoring.CheckPortAvailability(zipkinRemotePort); e != nil {
+			err = fmt.Errorf("Zipkin port not available on the machine: %w", err)
 			return
 		}
 
-		zipkinPods, err := monitoring.GetPods(kubeClientset, app, istioNS)
-		if err != nil {
-			logf("Error retrieving Zipkin pod details: %v", err)
+		zipkinPods, e := monitoring.GetPods(kubeClientset, appLabel, zipkinNamespace)
+		if e != nil {
+			err = fmt.Errorf("error retrieving Zipkin pod details: %w", err)
 			return
 		}
 
-		zipkinPortForwardPID, err = monitoring.PortForward(logf, zipkinPods, ZipkinPort, ZipkinPort, istioNS)
-		if err != nil {
-			logf("Error starting kubectl port-forward command: %v", err)
+		zipkinPortForwardPID, e = monitoring.PortForward(logf, zipkinPods, ZipkinPort, zipkinRemotePort, zipkinNamespace)
+		if e != nil {
+			err = fmt.Errorf("error starting kubectl port-forward command: %w", err)
 			return
 		}
 
@@ -93,9 +137,15 @@ func SetupZipkinTracing(kubeClientset *kubernetes.Clientset, logf logging.Format
 		// Applying AlwaysSample config to ensure we propagate zipkin header for every request made by this client.
 		trace.ApplyConfig(trace.Config{DefaultSampler: trace.AlwaysSample()})
 		logf("Successfully setup SpoofingClient for Zipkin Tracing")
-		ZipkinTracingEnabled = true
 	})
-	return ZipkinTracingEnabled
+	return
+}
+
+// SetupZipkinTracingOrFail is same as SetupZipkinTracing, but fails the test if an error happens
+func SetupZipkinTracingOrFail(t testing.TB, kubeClientset *kubernetes.Clientset, zipkinRemotePort int, zipkinNamespace string) {
+	if err := SetupZipkinTracing(kubeClientset, t.Logf, zipkinRemotePort, zipkinNamespace); err != nil {
+		t.Fatalf("Error while setup zipkin tracing: %v", err)
+	}
 }
 
 // CleanupZipkinTracingSetup cleans up the Zipkin tracing setup on the machine. This involves killing the process performing port-forward.
@@ -187,3 +237,11 @@ func jsonTrace(traceID string) ([]model.SpanModel, error) {
 	}
 	return models, nil
 }
+
+func parseNamespaceFromHostname(hostname string) (string, error) {
+	parts := strings.Split(hostname, ".")
+	if len(parts) < 3 || parts[2] != "svc" {
+		return "", fmt.Errorf("could not extract namespace/name from %s", hostname)
+	}
+	return parts[1], nil
+}