add apiserver tracing integration test, and fix endpoint validation

go.mod

@@ -83,7 +83,9 @@ require (
 	github.com/vmware/govmomi v0.20.3
 	go.etcd.io/etcd/client/pkg/v3 v3.5.0
 	go.etcd.io/etcd/client/v3 v3.5.0
+	go.opentelemetry.io/otel/sdk v0.20.0
 	go.opentelemetry.io/otel/trace v0.20.0
+	go.opentelemetry.io/proto/otlp v0.7.0
 	golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83
 	golang.org/x/exp v0.0.0-20210220032938-85be41e4509f // indirect
 	golang.org/x/net v0.0.0-20210520170846-37e1c6afe023

staging/src/k8s.io/apiserver/pkg/tracing/config.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/url"
+	"strings"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
@@ -93,23 +94,26 @@ func validateSamplingRate(rate int32, fldPath *field.Path) field.ErrorList {
 
 func validateEndpoint(endpoint string, fldPath *field.Path) field.ErrorList {
 	errs := field.ErrorList{}
-	if url, err := url.Parse(endpoint); err != nil {
+	if !strings.Contains(endpoint, "//") {
+		endpoint = "dns://" + endpoint
+	}
+	url, err := url.Parse(endpoint)
+	if err != nil {
 		errs = append(errs, field.Invalid(
 			fldPath, endpoint,
 			err.Error(),
 		))
-		// If the Host is empty, it indicates no scheme was specified, which is valid.
+		return errs
-	} else if url.Host != "" {
+	}
-		switch url.Scheme {
+	switch url.Scheme {
-		case "dns":
+	case "dns":
-		case "unix":
+	case "unix":
-		case "unix-abstract":
+	case "unix-abstract":
-		default:
+	default:
-			errs = append(errs, field.Invalid(
+		errs = append(errs, field.Invalid(
-				fldPath, endpoint,
+			fldPath, endpoint,
-				fmt.Sprintf("unsupported scheme: %v.  Options are none, dns, unix, or unix-abstract.  See https://github.com/grpc/grpc/blob/master/doc/naming.md", url.Scheme),
+			fmt.Sprintf("unsupported scheme: %v.  Options are none, dns, unix, or unix-abstract.  See https://github.com/grpc/grpc/blob/master/doc/naming.md", url.Scheme),
-			))
+		))
-		}
 	}
 	return errs
 }

staging/src/k8s.io/apiserver/pkg/tracing/config_test.go

@@ -30,6 +30,7 @@ import (
 
 var (
 	localhost    = "localhost:4317"
+	ipAddress    = "127.0.0.1:4317"
 	samplingRate = int32(12345)
 )
 
@@ -78,6 +79,23 @@ samplingRatePerMillion: 12345
 			},
 			expectedError: nil,
 		},
+		{
+			name:       "ip address",
+			createFile: true,
+			contents: `
+apiVersion: apiserver.config.k8s.io/v1alpha1
+kind: TracingConfiguration
+endpoint: 127.0.0.1:4317
+`,
+			expectedResult: &apiserver.TracingConfiguration{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "",
+					APIVersion: "",
+				},
+				Endpoint: &ipAddress,
+			},
+			expectedError: nil,
+		},
 		{
 			name:       "wrong_type",
 			createFile: true,

test/integration/apiserver/tracing/OWNERS

@@ -0,0 +1,3 @@
+# See the OWNERS docs at https://go.k8s.io/owners
+reviewers:
+- sig-instrumentation-reviewers

test/integration/apiserver/tracing/main_test.go

@@ -0,0 +1,27 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tracing
+
+import (
+	"testing"
+
+	"k8s.io/kubernetes/test/integration/framework"
+)
+
+func TestMain(m *testing.M) {
+	framework.EtcdMain(m.Run)
+}

test/integration/apiserver/tracing/tracing_test.go

@@ -0,0 +1,142 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tracing
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/trace"
+	traceservice "go.opentelemetry.io/proto/otlp/collector/trace/v1"
+	"google.golang.org/grpc"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	genericfeatures "k8s.io/apiserver/pkg/features"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	client "k8s.io/client-go/kubernetes"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/component-base/traces"
+	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
+	"k8s.io/kubernetes/test/integration/framework"
+)
+
+func TestAPIServerTracing(t *testing.T) {
+	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, genericfeatures.APIServerTracing, true)()
+
+	// Listen for traces from the API Server before starting it, so the
+	// API Server will successfully connect right away during the test.
+	listener, err := net.Listen("tcp", "localhost:")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	traceFound := make(chan struct{})
+	defer close(traceFound)
+	srv := grpc.NewServer()
+	traceservice.RegisterTraceServiceServer(srv, &traceServer{
+		traceFound: traceFound,
+		filterFunc: containsNodeListSpan})
+
+	go srv.Serve(listener)
+	defer srv.Stop()
+
+	// Write the configuration for tracing to a file
+	tracingConfigFile, err := ioutil.TempFile("", "tracing-config.yaml")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.Remove(tracingConfigFile.Name())
+
+	if err := ioutil.WriteFile(tracingConfigFile.Name(), []byte(fmt.Sprintf(`
+apiVersion: apiserver.config.k8s.io/v1alpha1
+kind: TracingConfiguration
+endpoint: %s`, listener.Addr().String())), os.FileMode(0755)); err != nil {
+		t.Fatal(err)
+	}
+
+	// Start the API Server with our tracing configuration
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	testServer := kubeapiservertesting.StartTestServerOrDie(t,
+		kubeapiservertesting.NewDefaultTestServerOptions(),
+		[]string{"--tracing-config-file=" + tracingConfigFile.Name()},
+		framework.SharedEtcd(),
+	)
+	clientConfig := testServer.ClientConfig
+
+	// Create a client that creates sampled traces.
+	tp := trace.TracerProvider(sdktrace.NewTracerProvider(sdktrace.WithSampler(sdktrace.AlwaysSample())))
+	clientConfig.Wrap(traces.WrapperFor(&tp))
+	clientSet, err := client.NewForConfig(clientConfig)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Make a request with the instrumented client
+	_, err = clientSet.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait for a span to be recorded from our request
+	select {
+	case <-traceFound:
+		return
+	case <-time.After(30 * time.Second):
+		t.Fatal("Timed out waiting for trace")
+	}
+}
+
+func containsNodeListSpan(req *traceservice.ExportTraceServiceRequest) bool {
+	for _, resourceSpans := range req.GetResourceSpans() {
+		for _, instrumentationSpans := range resourceSpans.GetInstrumentationLibrarySpans() {
+			for _, span := range instrumentationSpans.GetSpans() {
+				if span.Name != "KubernetesAPI" {
+					continue
+				}
+				for _, attr := range span.GetAttributes() {
+					if attr.GetKey() == "http.target" && attr.GetValue().GetStringValue() == "/api/v1/nodes" {
+						// We found our request!
+						return true
+					}
+				}
+			}
+		}
+	}
+	return false
+}
+
+// traceServer implements TracesServiceServer
+type traceServer struct {
+	traceFound chan struct{}
+	filterFunc func(req *traceservice.ExportTraceServiceRequest) bool
+	traceservice.UnimplementedTraceServiceServer
+}
+
+func (t *traceServer) Export(ctx context.Context, req *traceservice.ExportTraceServiceRequest) (*traceservice.ExportTraceServiceResponse, error) {
+	var emptyValue = traceservice.ExportTraceServiceResponse{}
+	if t.filterFunc(req) {
+		t.traceFound <- struct{}{}
+	}
+	return &emptyValue, nil
+}

vendor/modules.txt

@@ -909,6 +909,7 @@ go.opentelemetry.io/otel/metric/global
 go.opentelemetry.io/otel/metric/number
 go.opentelemetry.io/otel/metric/registry
 # go.opentelemetry.io/otel/sdk v0.20.0 => go.opentelemetry.io/otel/sdk v0.20.0
+## explicit
 go.opentelemetry.io/otel/sdk/instrumentation
 go.opentelemetry.io/otel/sdk/internal
 go.opentelemetry.io/otel/sdk/resource
@@ -932,6 +933,7 @@ go.opentelemetry.io/otel/sdk/metric/selector/simple
 ## explicit
 go.opentelemetry.io/otel/trace
 # go.opentelemetry.io/proto/otlp v0.7.0 => go.opentelemetry.io/proto/otlp v0.7.0
+## explicit
 go.opentelemetry.io/proto/otlp/collector/metrics/v1
 go.opentelemetry.io/proto/otlp/collector/trace/v1
 go.opentelemetry.io/proto/otlp/common/v1