Bump go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc from 0.42.0 to 0.46.0 (#1176)

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/ci.yml

@@ -9,37 +9,43 @@ on:
     branches: [master]
   pull_request:
 
+permissions:
+  contents: read
+
 env:
-  GO_VERSION: "1.18"
+  GO_VERSION: "1.20"
 
 jobs:
   lint:
     name: Lint
-    runs-on: ubuntu-latest
-    timeout-minutes: 10
+    runs-on: ubuntu-20.04
+    timeout-minutes: 20
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
-      - uses: actions/setup-go@v3
+        uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
         with:
           go-version: ${{ env.GO_VERSION }}
-      - uses: actions/checkout@v3
+          cache: false
+      - uses: actions/checkout@v4
       - uses: golangci/golangci-lint-action@v3
         with:
-          version: v1.48.0
-          args: --timeout=5m
+          version: v1.51.0
+          args: --timeout=15m --config=.golangci.yml
+          skip-cache: true
+          skip-build-cache: true
 
   unit-tests:
     name: Unit Tests
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 20
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
-      - uses: actions/setup-go@v3
+        uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
         with:
           go-version: ${{ env.GO_VERSION }}
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Run Tests
         run: make test
 
@@ -49,18 +55,18 @@ jobs:
     timeout-minutes: 10
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
-      - uses: actions/setup-go@v3
+        uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
         with:
           go-version: ${{ env.GO_VERSION }}
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Run Tests
         run: make envtest
 
   e2e:
     name: E2E
     runs-on: ubuntu-22.04
-    timeout-minutes: 10
+    timeout-minutes: 20
     env:
       CHANGE_MINIKUBE_NONE_USER: true
       KUBERNETES_VERSION: v1.20.1
@@ -72,11 +78,11 @@ jobs:
       GO111MODULE: "on"
 
     steps:
-      - uses: actions/setup-go@v3
+      - uses: actions/setup-go@v5
         with:
           go-version: ${{ env.GO_VERSION }}
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
       - name: Install Skaffold
         run: |
           curl -Lo skaffold https://storage.googleapis.com/skaffold/releases/${SKAFFOLD_VERSION}/skaffold-linux-amd64

.github/workflows/codeql-analysis.yml

@@ -27,7 +27,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       # Initializes the CodeQL tools for scanning.
       - name: Initialize CodeQL

.gitignore

@@ -41,3 +41,5 @@ loganalytics.json
 **/terraform-provider-kubernetes
 **/*.tfstate*
 debug
+
+vendor/

Dockerfile

@@ -1,6 +1,6 @@
 ARG GOLANG_CI_LINT_VERSION
 
-FROM golang:1.18 as builder
+FROM golang:1.20 as builder
 ENV PATH /go/bin:/usr/local/go/bin:$PATH
 ENV GOPATH /go
 COPY . /go/src/github.com/virtual-kubelet/virtual-kubelet

Makefile

@@ -184,7 +184,7 @@ fmt:
 	goimports -w $(shell go list -f '{{.Dir}}' ./...)
 
 
-export GOLANG_CI_LINT_VERSION ?= v1.48.0
+export GOLANG_CI_LINT_VERSION ?= v1.49.0
 DOCKER_BUILD ?= docker buildx build
 
 .PHONY: lint

Makefile.e2e

@@ -1,7 +1,17 @@
+
+# skaffold checks for kubectl context
+# For minikube, docker-for-desktop and docker-desktop the context matches above names
+# If one wants to use kind, kind gives context names based on the cluster-name
+# But as of now they match the following syntax: kind-*
+# The first check verifies that this is a kind kubernetes context
+# Second check verifies the other ones
 .PHONY: skaffold.validate
 skaffold.validate: kubectl_context := $(shell kubectl config current-context)
 skaffold.validate:
-	@if [[ ! "minikube,docker-for-desktop,docker-desktop" =~ .*"$(kubectl_context)".* ]]; then \
+
+	@if [[ "$(kubectl_context)" =~ .*"kind".* ]]; then \
+		true; \
+	elif [[ ! "minikube,docker-for-desktop,docker-desktop" =~ .*"$(kubectl_context)".* ]]; then \
 		echo current-context is [$(kubectl_context)]. Must be one of [minikube,docker-for-desktop,docker-desktop]; \
 		false; \
 	fi

README.md

@@ -1,5 +1,7 @@
 # Virtual Kubelet
 
+[![Go Reference](https://pkg.go.dev/badge/github.com/virtual-kubelet/virtual-kubelet.svg)](https://pkg.go.dev/github.com/virtual-kubelet/virtual-kubelet)
+
 Virtual Kubelet is an open source [Kubernetes kubelet](https://kubernetes.io/docs/reference/generated/kubelet/)
 implementation that masquerades as a kubelet for the purposes of connecting Kubernetes to other APIs.
 This allows the nodes to be backed by other services like ACI, AWS Fargate, [IoT Edge](https://github.com/Azure/iot-edge-virtual-kubelet-provider), [Tensile Kube](https://github.com/virtual-kubelet/tensile-kube) etc. The primary scenario for VK is enabling the extension of the Kubernetes API into serverless container platforms like ACI and Fargate, though we are open to others. However, it should be noted that VK is explicitly not intended to be an alternative to Kubernetes federation.
@@ -169,7 +171,7 @@ performing the neccessary actions.
 
 There are 3 main interfaces:
 
-#### PodLifecylceHandler
+#### PodLifecycleHandler
 
 When pods are created, updated, or deleted from Kubernetes, these methods are
 called to handle those actions.
@@ -305,7 +307,7 @@ Enable the ServiceNodeExclusion flag, by modifying the Controller Manager manife
 Virtual Kubelet follows the [CNCF Code of Conduct](https://github.com/cncf/foundation/blob/master/code-of-conduct.md).
 Sign the [CNCF CLA](https://github.com/kubernetes/community/blob/master/CLA.md) to be able to make Pull Requests to this repo.
 
-Monthly Virtual Kubelet Office Hours are held at 10am PST on the last Thursday of every month in this [zoom meeting room](https://zoom.us/j/94701509915). Check out the calendar [here](https://calendar.google.com/calendar?cid=bjRtbGMxYWNtNXR0NXQ1a2hqZmRkNTRncGNAZ3JvdXAuY2FsZW5kYXIuZ29vZ2xlLmNvbQ).
+Monthly Virtual Kubelet Office Hours are held at 10am PST on the second Thursday of every month in this [zoom meeting room](https://zoom.us/j/94701509915). Check out the calendar [here](https://calendar.google.com/calendar/embed?src=b119ced62134053de07d6c261b50d21ebde0da54f4163f5771b60ecf906e8b90%40group.calendar.google.com&ctz=America%2FLos_Angeles).
 
 Our google drive with design specifications and meeting notes are [here](https://drive.google.com/drive/folders/19Ndu11WBCCBDowo9CrrGUHoIfd2L8Ueg?usp=sharing).
 

cmd/virtual-kubelet/internal/commands/root/root.go

@@ -73,6 +73,13 @@ func runRootCommand(ctx context.Context, s *provider.Store, c Opts) error {
 		}
 	}
 
+	// Ensure API client.
+	clientSet, err := nodeutil.ClientsetFromEnv(c.KubeConfigPath)
+	if err != nil {
+		return err
+	}
+
+	// Set-up the node provider.
 	mux := http.NewServeMux()
 	newProvider := func(cfg nodeutil.ProviderConfig) (nodeutil.Provider, node.NodeProvider, error) {
 		rm, err := manager.NewResourceManager(cfg.Pods, cfg.Secrets, cfg.ConfigMaps, cfg.Services)
@@ -127,6 +134,7 @@ func runRootCommand(ctx context.Context, s *provider.Store, c Opts) error {
 
 		return nil
 	},
+		nodeutil.WithClient(clientSet),
 		setAuth(c.NodeName, apiConfig),
 		nodeutil.WithTLSConfig(
 			nodeutil.WithKeyPairFromPath(apiConfig.CertPath, apiConfig.KeyPath),

cmd/virtual-kubelet/internal/provider/mock/mock.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"time"
 
+	dto "github.com/prometheus/client_model/go"
 	"github.com/virtual-kubelet/virtual-kubelet/errdefs"
 	"github.com/virtual-kubelet/virtual-kubelet/log"
 	"github.com/virtual-kubelet/virtual-kubelet/node/api"
@@ -55,9 +56,11 @@ type MockProvider struct { //nolint:golint
 
 // MockConfig contains a mock virtual-kubelet's configurable parameters.
 type MockConfig struct { //nolint:golint
-	CPU    string `json:"cpu,omitempty"`
-	Memory string `json:"memory,omitempty"`
-	Pods   string `json:"pods,omitempty"`
+	CPU        string            `json:"cpu,omitempty"`
+	Memory     string            `json:"memory,omitempty"`
+	Pods       string            `json:"pods,omitempty"`
+	Others     map[string]string `json:"others,omitempty"`
+	ProviderID string            `json:"providerID,omitempty"`
 }
 
 // NewMockProviderMockConfig creates a new MockV0Provider. Mock legacy provider does not implement the new asynchronous podnotifier interface
@@ -128,6 +131,11 @@ func loadConfig(providerConfig, nodeName string) (config MockConfig, err error)
 	if _, err = resource.ParseQuantity(config.Pods); err != nil {
 		return config, fmt.Errorf("Invalid pods value %v", config.Pods)
 	}
+	for _, v := range config.Others {
+		if _, err = resource.ParseQuantity(v); err != nil {
+			return config, fmt.Errorf("Invalid other value %v", v)
+		}
+	}
 	return config, nil
 }
 
@@ -293,6 +301,19 @@ func (p *MockProvider) RunInContainer(ctx context.Context, namespace, name, cont
 	return nil
 }
 
+// AttachToContainer attaches to the executing process of a container in the pod, copying data
+// between in/out/err and the container's stdin/stdout/stderr.
+func (p *MockProvider) AttachToContainer(ctx context.Context, namespace, name, container string, attach api.AttachIO) error {
+	log.G(ctx).Infof("receive AttachToContainer %q", container)
+	return nil
+}
+
+// PortForward forwards a local port to a port on the pod
+func (p *MockProvider) PortForward(ctx context.Context, namespace, pod string, port int32, stream io.ReadWriteCloser) error {
+	log.G(ctx).Infof("receive PortForward %q", pod)
+	return nil
+}
+
 // GetPodStatus returns the status of a pod by name that is "running".
 // returns nil if a pod by that name is not found.
 func (p *MockProvider) GetPodStatus(ctx context.Context, namespace, name string) (*v1.PodStatus, error) {
@@ -332,6 +353,9 @@ func (p *MockProvider) ConfigureNode(ctx context.Context, n *v1.Node) { //nolint
 	ctx, span := trace.StartSpan(ctx, "mock.ConfigureNode") //nolint:staticcheck,ineffassign
 	defer span.End()
 
+	if p.config.ProviderID != "" {
+		n.Spec.ProviderID = p.config.ProviderID
+	}
 	n.Status.Capacity = p.capacity()
 	n.Status.Allocatable = p.capacity()
 	n.Status.Conditions = p.nodeConditions()
@@ -349,11 +373,15 @@ func (p *MockProvider) ConfigureNode(ctx context.Context, n *v1.Node) { //nolint
 
 // Capacity returns a resource list containing the capacity limits.
 func (p *MockProvider) capacity() v1.ResourceList {
-	return v1.ResourceList{
+	rl := v1.ResourceList{
 		"cpu":    resource.MustParse(p.config.CPU),
 		"memory": resource.MustParse(p.config.Memory),
 		"pods":   resource.MustParse(p.config.Pods),
 	}
+	for k, v := range p.config.Others {
+		rl[v1.ResourceName(k)] = resource.MustParse(v)
+	}
+	return rl
 }
 
 // NodeConditions returns a list of conditions (Ready, OutOfDisk, etc), for updates to the node status
@@ -508,6 +536,129 @@ func (p *MockProvider) GetStatsSummary(ctx context.Context) (*stats.Summary, err
 	return res, nil
 }
 
+func (p *MockProvider) generateMockMetrics(metricsMap map[string][]*dto.Metric, resourceType string, label []*dto.LabelPair) map[string][]*dto.Metric {
+	var (
+		cpuMetricSuffix    = "_cpu_usage_seconds_total"
+		memoryMetricSuffix = "_memory_working_set_bytes"
+		dummyValue         = float64(100)
+	)
+
+	if metricsMap == nil {
+		metricsMap = map[string][]*dto.Metric{}
+	}
+
+	finalCpuMetricName := resourceType + cpuMetricSuffix
+	finalMemoryMetricName := resourceType + memoryMetricSuffix
+
+	newCPUMetric := dto.Metric{
+		Label: label,
+		Counter: &dto.Counter{
+			Value: &dummyValue,
+		},
+	}
+	newMemoryMetric := dto.Metric{
+		Label: label,
+		Gauge: &dto.Gauge{
+			Value: &dummyValue,
+		},
+	}
+	// if metric family exists add to metric array
+	if cpuMetrics, ok := metricsMap[finalCpuMetricName]; ok {
+		metricsMap[finalCpuMetricName] = append(cpuMetrics, &newCPUMetric)
+	} else {
+		metricsMap[finalCpuMetricName] = []*dto.Metric{&newCPUMetric}
+	}
+	if memoryMetrics, ok := metricsMap[finalMemoryMetricName]; ok {
+		metricsMap[finalMemoryMetricName] = append(memoryMetrics, &newMemoryMetric)
+	} else {
+		metricsMap[finalMemoryMetricName] = []*dto.Metric{&newMemoryMetric}
+	}
+
+	return metricsMap
+}
+
+func (p *MockProvider) getMetricType(metricName string) *dto.MetricType {
+	var (
+		dtoCounterMetricType = dto.MetricType_COUNTER
+		dtoGaugeMetricType   = dto.MetricType_GAUGE
+		cpuMetricSuffix      = "_cpu_usage_seconds_total"
+		memoryMetricSuffix   = "_memory_working_set_bytes"
+	)
+	if strings.HasSuffix(metricName, cpuMetricSuffix) {
+		return &dtoCounterMetricType
+	}
+	if strings.HasSuffix(metricName, memoryMetricSuffix) {
+		return &dtoGaugeMetricType
+	}
+
+	return nil
+}
+
+func (p *MockProvider) GetMetricsResource(ctx context.Context) ([]*dto.MetricFamily, error) {
+	var span trace.Span
+	ctx, span = trace.StartSpan(ctx, "GetMetricsResource") //nolint: ineffassign,staticcheck
+	defer span.End()
+
+	var (
+		nodeNameStr      = "NodeName"
+		podNameStr       = "PodName"
+		containerNameStr = "containerName"
+	)
+	nodeLabels := []*dto.LabelPair{
+		{
+			Name:  &nodeNameStr,
+			Value: &p.nodeName,
+		},
+	}
+
+	metricsMap := p.generateMockMetrics(nil, "node", nodeLabels)
+	for _, pod := range p.pods {
+		podLabels := []*dto.LabelPair{
+			{
+				Name:  &nodeNameStr,
+				Value: &p.nodeName,
+			},
+			{
+				Name:  &podNameStr,
+				Value: &pod.Name,
+			},
+		}
+		metricsMap = p.generateMockMetrics(metricsMap, "pod", podLabels)
+		for _, container := range pod.Spec.Containers {
+			containerLabels := []*dto.LabelPair{
+				{
+					Name:  &nodeNameStr,
+					Value: &p.nodeName,
+				},
+				{
+					Name:  &podNameStr,
+					Value: &pod.Name,
+				},
+				{
+					Name:  &containerNameStr,
+					Value: &container.Name,
+				},
+			}
+			metricsMap = p.generateMockMetrics(metricsMap, "container", containerLabels)
+		}
+	}
+
+	res := []*dto.MetricFamily{}
+	for metricName := range metricsMap {
+		tempName := metricName
+		tempMetrics := metricsMap[tempName]
+
+		metricFamily := dto.MetricFamily{
+			Name:   &tempName,
+			Type:   p.getMetricType(tempName),
+			Metric: tempMetrics,
+		}
+		res = append(res, &metricFamily)
+	}
+
+	return res, nil
+}
+
 // NotifyPods is called to set a pod notifier callback function. This should be called before any operations are done
 // within the provider.
 func (p *MockProvider) NotifyPods(ctx context.Context, notifier func(*v1.Pod)) {

docs/proposals/MetricsUpdateProposal.md

@@ -0,0 +1,296 @@
+#  Virtual Kubelet Metrics Update
+
+<!-- toc -->
+- [Summary](#summary)
+- [Motivation](#motivation)
+  - [Goals](#goals)
+  - [Non-Goals](#non-goals)
+- [Proposal](#proposal)
+- [Design Details](#design-details)
+  - [API](#api)
+  - [Data](#data)
+  - [Changes to the Provider](#changes-to-the-provider)
+  - [Test Plan](#test-plan)
+<!-- /toc -->
+
+## Summary
+
+Add the new /metrics/resource endpoint in the virtual-kubelet to support the metrics server update for new Kubernetes versions `>=1.24`
+
+
+## Motivation
+
+The Kubernetes metrics server now tries to get metrics from the kubelet using the new metrics endpoint [/metrics/resource](https://github.com/kubernetes-sigs/metrics-server/commit/a2d732e5cdbfd93a6ebce221e8df0e8b463eecc6#diff-6e5b914d1403a14af1cc43582a2c9af727113037a3c6a77d8729aaefba084fb5R88),
+while Virtual Kubelet is still exposing the earlier metrics endpoint [/stats/summary](https://github.com/virtual-kubelet/virtual-kubelet/blob/master/node/api/server.go#L90). 
+This causes metrics to break when using virtual kubelet with newer Kubernetes versions (>=1.24). 
+To support the new metrics server, this document proposes adding a new handler to handle the updated metrics endpoint. 
+This will be an additive update, and the old 
+[/stats/summary](https://github.com/virtual-kubelet/virtual-kubelet/blob/master/node/api/server.go#L90) endpoint will still be available to maintain backward compatibility with 
+the older metrics server version.
+
+
+### Goals
+
+- Support metrics for kubernetes version `>=1.24` through adding /metrics/resource endpoint handler.
+
+### Non-Goals
+
+- Ensure pod autoscaling works as expected with the newer kubernetes versions `>=1.24` as expected
+
+## Proposal
+
+Add a new handler for `/metrics/resource` endpoint that calls a new `GetMetricsResource` method in the provider,
+which in-turn returns metrics using the prometheus `model.Samples` data structure as expected by the new metrics server.
+The provider will need to implement the `GetMetricsResource` method in order to add support for the new `/metrics/resource` endpoint with Kubernetes version >=1.24
+
+
+## Design Details
+Currently the virtual kubelet code uses the `PodStatsSummaryHandler` method to set up a http handler for serving pod metrics via the `/stats/summary` endpoint. 
+To support the updated metrics server, we need to add another handler `PodMetricsResourceHandler`  which can serve metrics via the `/metrics/resource` endpoint. 
+The `PodMetricsResourceHandler` calls the new `GetMetricsResource` method of the provider to get the metrics from the specific provider. 
+
+### API
+Add `GetMetricsResource` to `PodHandlerConfig`
+```go
+type PodHandlerConfig struct { //nolint:golint
+	RunInContainer   ContainerExecHandlerFunc
+	GetContainerLogs ContainerLogsHandlerFunc
+	// GetPods is meant to enumerate the pods that the provider knows about
+	GetPods PodListerFunc
+	// GetPodsFromKubernetes is meant to enumerate the pods that the node is meant to be running
+	GetPodsFromKubernetes PodListerFunc
+	GetStatsSummary       PodStatsSummaryHandlerFunc
+	GetMetricsResource    PodMetricsResourceHandlerFunc
+	StreamIdleTimeout     time.Duration
+	StreamCreationTimeout time.Duration
+}
+```
+Add endpoint to `PodHandler` method 
+```go
+const MetricsResourceRouteSuffix = "/metrics/resource"
+
+func PodHandler(p PodHandlerConfig, debug bool) http.Handler {
+	r := mux.NewRouter()
+
+	// This matches the behaviour in the reference kubelet
+	r.StrictSlash(true)
+	if debug {
+		r.HandleFunc("/runningpods/", HandleRunningPods(p.GetPods)).Methods("GET")
+	}
+
+	r.HandleFunc("/pods", HandleRunningPods(p.GetPodsFromKubernetes)).Methods("GET")
+	r.HandleFunc("/containerLogs/{namespace}/{pod}/{container}", HandleContainerLogs(p.GetContainerLogs)).Methods("GET")
+	r.HandleFunc(
+		"/exec/{namespace}/{pod}/{container}",
+		HandleContainerExec(
+			p.RunInContainer,
+			WithExecStreamCreationTimeout(p.StreamCreationTimeout),
+			WithExecStreamIdleTimeout(p.StreamIdleTimeout),
+		),
+	).Methods("POST", "GET")
+
+	if p.GetStatsSummary != nil {
+		f := HandlePodStatsSummary(p.GetStatsSummary)
+		r.HandleFunc("/stats/summary", f).Methods("GET")
+		r.HandleFunc("/stats/summary/", f).Methods("GET")
+	}
+
+	if p.GetMetricsResource != nil {
+		f := HandlePodMetricsResource(p.GetMetricsResource)
+		r.HandleFunc(MetricsResourceRouteSuffix, f).Methods("GET")
+		r.HandleFunc(MetricsResourceRouteSuffix+"/", f).Methods("GET")
+	}
+	r.NotFoundHandler = http.HandlerFunc(NotFound)
+	return r
+}
+```
+  
+New `PodMetricsResourceHandler` method, that uses the new `PodMetricsResourceHandlerFunc` definition.
+```go
+// PodMetricsResourceHandler creates an http handler for serving pod metrics.
+//
+// If the passed in handler func is nil this will create handlers which only
+// serves http.StatusNotImplemented
+func PodMetricsResourceHandler(f PodMetricsResourceHandlerFunc) http.Handler {
+	if f == nil {
+		return http.HandlerFunc(NotImplemented)
+	}
+
+	r := mux.NewRouter()
+
+	h := HandlePodMetricsResource(f)
+
+	r.Handle(MetricsResourceRouteSuffix, ochttp.WithRouteTag(h, "PodMetricsResourceHandler")).Methods("GET")
+	r.Handle(MetricsResourceRouteSuffix+"/", ochttp.WithRouteTag(h, "PodMetricsResourceHandler")).Methods("GET")
+
+	r.NotFoundHandler = http.HandlerFunc(NotFound)
+	return r
+}
+
+```
+ 
+ 
+`HandlePodMetricsResource` method returns a HandlerFunc which serves the metrics encoded in prometheus' text format encoding as expected by the  metrics-server
+```go
+// HandlePodMetricsResource makes an HTTP handler for implementing the kubelet /metrics/resource endpoint
+func HandlePodMetricsResource(h PodMetricsResourceHandlerFunc) http.HandlerFunc {
+	if h == nil {
+		return NotImplemented
+	}
+	return handleError(func(w http.ResponseWriter, req *http.Request) error {
+		metrics, err := h(req.Context())
+		if err != nil {
+			if isCancelled(err) {
+				return err
+			}
+			return errors.Wrap(err, "error getting status from provider")
+		}
+
+		b, err := json.Marshal(metrics)
+		if err != nil {
+			return errors.Wrap(err, "error marshalling metrics")
+		}
+
+		if _, err := w.Write(b); err != nil {
+			return errors.Wrap(err, "could not write to client")
+		}
+		return nil
+	})
+}
+```
+ 
+The `PodMetricsResourceHandlerFunc` returns the metrics data using Prometheus' `MetricFamily` data structure. More details are provided in the Data subsection
+```go
+// PodMetricsResourceHandlerFunc defines the handler for getting pod metrics
+type PodMetricsResourceHandlerFunc func(context.Context) ([]*dto.MetricFamily, error)
+```
+ 
+### Data
+
+The updated metrics server does not add any new fields to the metrics data but uses the Prometheus textparse series parser to parse and reconstruct the [MetricsBatch](https://github.com/kubernetes-sigs/metrics-server/blob/83b2e01f9825849ae5f562e47aa1a4178b5d06e5/pkg/storage/types.go#L31) data structure.  
+Currently virtual-kubelet is sending data to the server using the [summary](https://github.com/virtual-kubelet/virtual-kubelet/blob/be0a062aec9a5eeea3ad6fbe5aec557a235558f6/node/api/statsv1alpha1/types.go#L24) data structure. The Prometheus text parser expects a series of bytes as in the Prometheus [model.Samples](https://github.com/kubernetes/kubernetes/blob/a93eda9db305611cacd8b6ee930ab3149a08f9b0/vendor/github.com/prometheus/common/model/value.go#L184) data structure, similar to the test [here](https://github.com/prometheus/prometheus/blob/c70d85baed260f6013afd18d6cd0ffcac4339861/model/textparse/promparse_test.go#L31).
+ 
+Examples of how the new metrics are defined may be seen in the Kubernetes e2e test that calls the /metrics/resource endpoint [here](https://github.com/kubernetes/kubernetes/blob/a93eda9db305611cacd8b6ee930ab3149a08f9b0/test/e2e_node/resource_metrics_test.go#L76), and the kubelet metrics defined in the Kubernetes/kubelet code [here](https://github.com/kubernetes/kubernetes/blob/master/pkg/kubelet/metrics/collectors/resource_metrics.go) .
+ 
+```go
+var (
+	nodeCPUUsageDesc = metrics.NewDesc("node_cpu_usage_seconds_total",
+		"Cumulative cpu time consumed by the node in core-seconds",
+		nil,
+		nil,
+		metrics.ALPHA,
+		"")
+
+	nodeMemoryUsageDesc = metrics.NewDesc("node_memory_working_set_bytes",
+		"Current working set of the node in bytes",
+		nil,
+		nil,
+		metrics.ALPHA,
+		"")
+
+	containerCPUUsageDesc = metrics.NewDesc("container_cpu_usage_seconds_total",
+		"Cumulative cpu time consumed by the container in core-seconds",
+		[]string{"container", "pod", "namespace"},
+		nil,
+		metrics.ALPHA,
+		"")
+
+	containerMemoryUsageDesc = metrics.NewDesc("container_memory_working_set_bytes",
+		"Current working set of the container in bytes",
+		[]string{"container", "pod", "namespace"},
+		nil,
+		metrics.ALPHA,
+		"")
+
+	podCPUUsageDesc = metrics.NewDesc("pod_cpu_usage_seconds_total",
+		"Cumulative cpu time consumed by the pod in core-seconds",
+		[]string{"pod", "namespace"},
+		nil,
+		metrics.ALPHA,
+		"")
+
+	podMemoryUsageDesc = metrics.NewDesc("pod_memory_working_set_bytes",
+		"Current working set of the pod in bytes",
+		[]string{"pod", "namespace"},
+		nil,
+		metrics.ALPHA,
+		"")
+
+	resourceScrapeResultDesc = metrics.NewDesc("scrape_error",
+		"1 if there was an error while getting container metrics, 0 otherwise",
+		nil,
+		nil,
+		metrics.ALPHA,
+		"")
+
+	containerStartTimeDesc = metrics.NewDesc("container_start_time_seconds",
+		"Start time of the container since unix epoch in seconds",
+		[]string{"container", "pod", "namespace"},
+		nil,
+		metrics.ALPHA,
+		"")
+)
+```
+ 
+The kubernetes/kubelet code implements Prometheus' [collector](https://github.com/kubernetes/kubernetes/blob/master/pkg/kubelet/metrics/collectors/resource_metrics.go#L88) interface which is used along with the k8s.io/component-base implementation of the [registry](https://github.com/kubernetes/component-base/blob/40d14bdbd62f9e2ea697f97d81d4abc72839901e/metrics/registry.go#L114) interface in order to collect and return the metrics data using the Prometheus' [MetricFamily](https://github.com/prometheus/client_model/blob/master/go/metrics.pb.go#L773) data structure.
+ 
+The Gather method in the registry calls the kubelet collector's Collect method, and returns the data using the MetricFamily data structure. The metrics server expects metrics to be encoded in prometheus'
+text format, and the kubelet uses the http handler from prometheus' promhttp module which returns the metrics data encoded in prometheus' text format encoding.
+```go
+type KubeRegistry interface {
+	// Deprecated
+	RawMustRegister(...prometheus.Collector)
+	// CustomRegister is our internal variant of Prometheus registry.Register
+	CustomRegister(c StableCollector) error
+	// CustomMustRegister is our internal variant of Prometheus registry.MustRegister
+	CustomMustRegister(cs ...StableCollector)
+	// Register conforms to Prometheus registry.Register
+	Register(Registerable) error
+	// MustRegister conforms to Prometheus registry.MustRegister
+	MustRegister(...Registerable)
+	// Unregister conforms to Prometheus registry.Unregister
+	Unregister(collector Collector) bool
+	// Gather conforms to Prometheus gatherer.Gather
+	Gather() ([]*dto.MetricFamily, error)
+	// Reset invokes the Reset() function on all items in the registry
+	// which are added as resettables.
+	Reset()
+}
+```
+
+Prometheus’ MetricsFamily data structure:  
+```go
+type MetricFamily struct {
+	Name                 *string     `protobuf:"bytes,1,opt,name=name" json:"name,omitempty"`
+	Help                 *string     `protobuf:"bytes,2,opt,name=help" json:"help,omitempty"`
+	Type                 *MetricType `protobuf:"varint,3,opt,name=type,enum=io.prometheus.client.MetricType" json:"type,omitempty"`
+	Metric               []*Metric   `protobuf:"bytes,4,rep,name=metric" json:"metric,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}    `json:"-"`
+	XXX_unrecognized     []byte      `json:"-"`
+	XXX_sizecache        int32       `json:"-"`
+}
+```
+
+Therefore the provider's GetMetricsResource method should use the same return type as the Gather method in the registry interface.
+ 
+### Changes to the Provider.
+
+In order to support the new metrics endpoint the Provider must implement the GetMetricsResource method with definition
+
+```golang
+
+import (
+  dto "github.com/prometheus/client_model/go"
+  "context"
+)
+
+func GetMetricsResource(context.Context) ([]*dto.MetricsFamily, error) {
+...
+}
+```
+
+### Test Plan
+
+- Write a provider implementation for GetMetricsResource method in ACI Provider and deploy pods get metrics using kubectl 
+- Run end-to-end tests with the provider implementation
+

go.mod

@@ -1,99 +1,121 @@
 module github.com/virtual-kubelet/virtual-kubelet
 
-go 1.17
+go 1.20
 
 require (
 	contrib.go.opencensus.io/exporter/jaeger v0.2.1
 	contrib.go.opencensus.io/exporter/ocagent v0.7.0
-	github.com/bombsimon/logrusr/v3 v3.0.0
-	github.com/elazarl/goproxy v0.0.0-20190421051319-9d40249d3c2f // indirect
-	github.com/google/go-cmp v0.5.8
+	github.com/bombsimon/logrusr/v3 v3.1.0
+	github.com/google/go-cmp v0.6.0
 	github.com/gorilla/mux v1.8.0
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.13.0
-	github.com/sirupsen/logrus v1.9.0
-	github.com/spf13/cobra v1.5.0
+	github.com/prometheus/client_golang v1.16.0
+	github.com/prometheus/client_model v0.4.0
+	github.com/prometheus/common v0.44.0
+	github.com/sirupsen/logrus v1.9.3
+	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
-	go.opencensus.io v0.23.0
-	go.opentelemetry.io/otel v0.20.0
-	go.opentelemetry.io/otel/sdk v0.20.0
-	go.opentelemetry.io/otel/trace v0.20.0
-	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
-	golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f
-	golang.org/x/time v0.0.0-20220609170525-579cf78fd858
+	github.com/stretchr/testify v1.8.4
+	go.opencensus.io v0.24.0
+	go.opentelemetry.io/otel v1.22.0
+	go.opentelemetry.io/otel/sdk v1.22.0
+	go.opentelemetry.io/otel/trace v1.22.0
+	golang.org/x/sync v0.5.0
+	golang.org/x/sys v0.16.0
+	golang.org/x/time v0.3.0
+	google.golang.org/protobuf v1.31.0
 	gotest.tools v2.2.0+incompatible
-	k8s.io/api v0.25.0
-	k8s.io/apimachinery v0.25.0
-	k8s.io/apiserver v0.25.0
-	k8s.io/client-go v0.25.0
-	k8s.io/klog/v2 v2.80.1
-	k8s.io/utils v0.0.0-20220728103510-ee6ede2d64ed
-	sigs.k8s.io/controller-runtime v0.13.0
+	k8s.io/api v0.29.1
+	k8s.io/apimachinery v0.29.1
+	k8s.io/apiserver v0.29.1
+	k8s.io/client-go v0.29.1
+	k8s.io/klog/v2 v2.110.1
+	k8s.io/utils v0.0.0-20230726121419-3b25d923346b
+	sigs.k8s.io/controller-runtime v0.15.0
 )
 
 require (
-	github.com/PuerkitoBio/purell v1.1.1 // indirect
-	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
+	github.com/NYTimes/gziphandler v1.1.1 // indirect
+	github.com/antlr/antlr4/runtime/Go/antlr/v4 v4.0.0-20230305170008-8188dc5388df // indirect
+	github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
-	github.com/census-instrumentation/opencensus-proto v0.2.1 // indirect
-	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
+	github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/coreos/go-semver v0.3.1 // indirect
+	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/emicklei/go-restful/v3 v3.8.0 // indirect
+	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/evanphx/json-patch v4.12.0+incompatible // indirect
 	github.com/evanphx/json-patch/v5 v5.6.0 // indirect
-	github.com/felixge/httpsnoop v1.0.1 // indirect
-	github.com/fsnotify/fsnotify v1.5.4 // indirect
-	github.com/go-logr/logr v1.2.3 // indirect
-	github.com/go-openapi/jsonpointer v0.19.5 // indirect
-	github.com/go-openapi/jsonreference v0.19.5 // indirect
-	github.com/go-openapi/swag v0.19.14 // indirect
+	github.com/felixge/httpsnoop v1.0.3 // indirect
+	github.com/fsnotify/fsnotify v1.7.0 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-openapi/jsonpointer v0.19.6 // indirect
+	github.com/go-openapi/jsonreference v0.20.2 // indirect
+	github.com/go-openapi/swag v0.22.3 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
-	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/google/gnostic v0.5.7-v3refs // indirect
-	github.com/google/gofuzz v1.1.0 // indirect
-	github.com/google/uuid v1.1.2 // indirect
-	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/google/cel-go v0.17.7 // indirect
+	github.com/google/gnostic-models v0.6.8 // indirect
+	github.com/google/gofuzz v1.2.0 // indirect
+	github.com/google/uuid v1.3.1 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
+	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect
 	github.com/imdario/mergo v0.3.12 // indirect
-	github.com/inconshreveable/mousetrap v1.0.0 // indirect
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
-	github.com/mailru/easyjson v0.7.6 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/moby/spdystream v0.2.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	github.com/prometheus/client_model v0.2.0 // indirect
-	github.com/prometheus/common v0.37.0 // indirect
-	github.com/prometheus/procfs v0.8.0 // indirect
+	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/procfs v0.10.1 // indirect
+	github.com/stoewer/go-strcase v1.2.0 // indirect
 	github.com/uber/jaeger-client-go v2.25.0+incompatible // indirect
-	go.opentelemetry.io/contrib v0.20.0 // indirect
-	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.20.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp v0.20.0 // indirect
-	go.opentelemetry.io/otel/metric v0.20.0 // indirect
-	go.opentelemetry.io/otel/sdk/export/metric v0.20.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v0.20.0 // indirect
-	go.opentelemetry.io/proto/otlp v0.7.0 // indirect
-	golang.org/x/net v0.0.0-20220722155237-a158d28d115b // indirect
-	golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b // indirect
-	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
-	golang.org/x/text v0.3.7 // indirect
-	google.golang.org/api v0.43.0 // indirect
+	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
+	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
+	go.etcd.io/etcd/client/v3 v3.5.10 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.44.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.19.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.19.0 // indirect
+	go.opentelemetry.io/otel/metric v1.22.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.24.0 // indirect
+	golang.org/x/crypto v0.17.0 // indirect
+	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect
+	golang.org/x/net v0.19.0 // indirect
+	golang.org/x/oauth2 v0.11.0 // indirect
+	golang.org/x/term v0.15.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	google.golang.org/api v0.30.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20220502173005-c8bf987b8c21 // indirect
-	google.golang.org/grpc v1.47.0 // indirect
-	google.golang.org/protobuf v1.28.1 // indirect
+	google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect
+	google.golang.org/grpc v1.59.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/apiextensions-apiserver v0.25.0 // indirect
-	k8s.io/component-base v0.25.0 // indirect
-	k8s.io/kube-openapi v0.0.0-20220803162953-67bda5d908f1 // indirect
-	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.32 // indirect
-	sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 // indirect
-	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
+	k8s.io/apiextensions-apiserver v0.27.2 // indirect
+	k8s.io/component-base v0.29.1 // indirect
+	k8s.io/kms v0.29.1 // indirect
+	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
+	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.28.0 // indirect
+	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
+	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect
 )

internal/kubernetes/portforward/constants.go

@@ -0,0 +1,24 @@
+/*
+Copyright 2015 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 portforward contains server-side logic for handling port forwarding requests.
+package portforward
+
+// ProtocolV1Name is the name of the subprotocol used for port forwarding.
+const ProtocolV1Name = "portforward.k8s.io"
+
+// SupportedProtocols are the supported port forwarding protocols.
+var SupportedProtocols = []string{ProtocolV1Name}

internal/kubernetes/portforward/httpstream.go

@@ -0,0 +1,317 @@
+/*
+Copyright 2016 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 portforward
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+	"sync"
+	"time"
+
+	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/httpstream"
+	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+
+	"k8s.io/klog/v2"
+)
+
+func handleHTTPStreams(req *http.Request, w http.ResponseWriter, portForwarder PortForwarder, podName string, uid types.UID, supportedPortForwardProtocols []string, idleTimeout, streamCreationTimeout time.Duration) error {
+	_, err := httpstream.Handshake(req, w, supportedPortForwardProtocols)
+	// negotiated protocol isn't currently used server side, but could be in the future
+	if err != nil {
+		// Handshake writes the error to the client
+		return err
+	}
+	streamChan := make(chan httpstream.Stream, 1)
+
+	klog.V(5).InfoS("Upgrading port forward response")
+	
+	// TODO aka-somix: SPDY is deprecated and it should be replaced in order to support HTTP/2
+	upgrader := spdy.NewResponseUpgrader()
+	conn := upgrader.UpgradeResponse(w, req, httpStreamReceived(streamChan))
+	if conn == nil {
+		return errors.New("unable to upgrade httpstream connection")
+	}
+	defer conn.Close()
+
+	klog.V(5).InfoS("Connection setting port forwarding streaming connection idle timeout", "connection", conn, "idleTimeout", idleTimeout)
+	conn.SetIdleTimeout(idleTimeout)
+
+	h := &httpStreamHandler{
+		conn:                  conn,
+		streamChan:            streamChan,
+		streamPairs:           make(map[string]*httpStreamPair),
+		streamCreationTimeout: streamCreationTimeout,
+		pod:                   podName,
+		uid:                   uid,
+		forwarder:             portForwarder,
+	}
+	h.run()
+
+	return nil
+}
+
+// httpStreamReceived is the httpstream.NewStreamHandler for port
+// forward streams. It checks each stream's port and stream type headers,
+// rejecting any streams that with missing or invalid values. Each valid
+// stream is sent to the streams channel.
+func httpStreamReceived(streams chan httpstream.Stream) func(httpstream.Stream, <-chan struct{}) error {
+	return func(stream httpstream.Stream, replySent <-chan struct{}) error {
+		// make sure it has a valid port header
+		portString := stream.Headers().Get(api.PortHeader)
+		if len(portString) == 0 {
+			return fmt.Errorf("%q header is required", api.PortHeader)
+		}
+		port, err := strconv.ParseUint(portString, 10, 16)
+		if err != nil {
+			return fmt.Errorf("unable to parse %q as a port: %v", portString, err)
+		}
+		if port < 1 {
+			return fmt.Errorf("port %q must be > 0", portString)
+		}
+
+		// make sure it has a valid stream type header
+		streamType := stream.Headers().Get(api.StreamType)
+		if len(streamType) == 0 {
+			return fmt.Errorf("%q header is required", api.StreamType)
+		}
+		if streamType != api.StreamTypeError && streamType != api.StreamTypeData {
+			return fmt.Errorf("invalid stream type %q", streamType)
+		}
+
+		streams <- stream
+		return nil
+	}
+}
+
+// httpStreamHandler is capable of processing multiple port forward
+// requests over a single httpstream.Connection.
+type httpStreamHandler struct {
+	conn                  httpstream.Connection
+	streamChan            chan httpstream.Stream
+	streamPairsLock       sync.RWMutex
+	streamPairs           map[string]*httpStreamPair
+	streamCreationTimeout time.Duration
+	pod                   string
+	uid                   types.UID
+	forwarder             PortForwarder
+}
+
+// getStreamPair returns a httpStreamPair for requestID. This creates a
+// new pair if one does not yet exist for the requestID. The returned bool is
+// true if the pair was created.
+func (h *httpStreamHandler) getStreamPair(requestID string) (*httpStreamPair, bool) {
+	h.streamPairsLock.Lock()
+	defer h.streamPairsLock.Unlock()
+
+	if p, ok := h.streamPairs[requestID]; ok {
+		klog.V(5).InfoS("Connection request found existing stream pair", "connection", h.conn, "request", requestID)
+		return p, false
+	}
+
+	klog.V(5).InfoS("Connection request creating new stream pair", "connection", h.conn, "request", requestID)
+
+	p := newPortForwardPair(requestID)
+	h.streamPairs[requestID] = p
+
+	return p, true
+}
+
+// monitorStreamPair waits for the pair to receive both its error and data
+// streams, or for the timeout to expire (whichever happens first), and then
+// removes the pair.
+func (h *httpStreamHandler) monitorStreamPair(p *httpStreamPair, timeout <-chan time.Time) {
+	select {
+	case <-timeout:
+		err := fmt.Errorf("(conn=%v, request=%s) timed out waiting for streams", h.conn, p.requestID)
+		utilruntime.HandleError(err)
+		p.printError(err.Error())
+	case <-p.complete:
+		klog.V(5).InfoS("Connection request successfully received error and data streams", "connection", h.conn, "request", p.requestID)
+	}
+	h.removeStreamPair(p.requestID)
+}
+
+// hasStreamPair returns a bool indicating if a stream pair for requestID
+// exists.
+func (h *httpStreamHandler) hasStreamPair(requestID string) bool {
+	h.streamPairsLock.RLock()
+	defer h.streamPairsLock.RUnlock()
+
+	_, ok := h.streamPairs[requestID]
+	return ok
+}
+
+// removeStreamPair removes the stream pair identified by requestID from streamPairs.
+func (h *httpStreamHandler) removeStreamPair(requestID string) {
+	h.streamPairsLock.Lock()
+	defer h.streamPairsLock.Unlock()
+
+	if h.conn != nil {
+		pair := h.streamPairs[requestID]
+		h.conn.RemoveStreams(pair.dataStream, pair.errorStream)
+	}
+	delete(h.streamPairs, requestID)
+}
+
+// requestID returns the request id for stream.
+func (h *httpStreamHandler) requestID(stream httpstream.Stream) string {
+	requestID := stream.Headers().Get(api.PortForwardRequestIDHeader)
+	if len(requestID) == 0 {
+		klog.V(5).InfoS("Connection stream received without requestID header", "connection", h.conn)
+		// If we get here, it's because the connection came from an older client
+		// that isn't generating the request id header
+		// (https://github.com/kubernetes/kubernetes/blob/843134885e7e0b360eb5441e85b1410a8b1a7a0c/pkg/client/unversioned/portforward/portforward.go#L258-L287)
+		//
+		// This is a best-effort attempt at supporting older clients.
+		//
+		// When there aren't concurrent new forwarded connections, each connection
+		// will have a pair of streams (data, error), and the stream IDs will be
+		// consecutive odd numbers, e.g. 1 and 3 for the first connection. Convert
+		// the stream ID into a pseudo-request id by taking the stream type and
+		// using id = stream.Identifier() when the stream type is error,
+		// and id = stream.Identifier() - 2 when it's data.
+		//
+		// NOTE: this only works when there are not concurrent new streams from
+		// multiple forwarded connections; it's a best-effort attempt at supporting
+		// old clients that don't generate request ids.  If there are concurrent
+		// new connections, it's possible that 1 connection gets streams whose IDs
+		// are not consecutive (e.g. 5 and 9 instead of 5 and 7).
+		streamType := stream.Headers().Get(api.StreamType)
+		switch streamType {
+		case api.StreamTypeError:
+			requestID = strconv.Itoa(int(stream.Identifier()))
+		case api.StreamTypeData:
+			requestID = strconv.Itoa(int(stream.Identifier()) - 2)
+		}
+
+		klog.V(5).InfoS("Connection automatically assigning request ID from stream type and stream ID", "connection", h.conn, "request", requestID, "streamType", streamType, "stream", stream.Identifier())
+	}
+	return requestID
+}
+
+// run is the main loop for the httpStreamHandler. It processes new
+// streams, invoking portForward for each complete stream pair. The loop exits
+// when the httpstream.Connection is closed.
+func (h *httpStreamHandler) run() {
+	klog.V(5).InfoS("Connection waiting for port forward streams", "connection", h.conn)
+Loop:
+	for {
+		select {
+		case <-h.conn.CloseChan():
+			klog.V(5).InfoS("Connection upgraded connection closed", "connection", h.conn)
+			break Loop
+		case stream := <-h.streamChan:
+			requestID := h.requestID(stream)
+			streamType := stream.Headers().Get(api.StreamType)
+			klog.V(5).InfoS("Connection request received new type of stream", "connection", h.conn, "request", requestID, "streamType", streamType)
+
+			p, created := h.getStreamPair(requestID)
+			if created {
+				go h.monitorStreamPair(p, time.After(h.streamCreationTimeout))
+			}
+			if complete, err := p.add(stream); err != nil {
+				msg := fmt.Sprintf("error processing stream for request %s: %v", requestID, err)
+				utilruntime.HandleError(errors.New(msg))
+				p.printError(msg)
+			} else if complete {
+				go h.portForward(p)
+			}
+		}
+	}
+}
+
+// portForward invokes the httpStreamHandler's forwarder.PortForward
+// function for the given stream pair.
+func (h *httpStreamHandler) portForward(p *httpStreamPair) {
+	ctx := context.Background()
+	defer p.dataStream.Close()
+	defer p.errorStream.Close()
+
+	portString := p.dataStream.Headers().Get(api.PortHeader)
+	port, _ := strconv.ParseInt(portString, 10, 32)
+
+	klog.V(5).InfoS("Connection request invoking forwarder.PortForward for port", "connection", h.conn, "request", p.requestID, "port", portString)
+	err := h.forwarder.PortForward(ctx, h.pod, h.uid, int32(port), p.dataStream)
+	klog.V(5).InfoS("Connection request done invoking forwarder.PortForward for port", "connection", h.conn, "request", p.requestID, "port", portString)
+
+	if err != nil {
+		msg := fmt.Errorf("error forwarding port %d to pod %s, uid %v: %v", port, h.pod, h.uid, err)
+		utilruntime.HandleError(msg)
+		fmt.Fprint(p.errorStream, msg.Error())
+	}
+}
+
+// httpStreamPair represents the error and data streams for a port
+// forwarding request.
+type httpStreamPair struct {
+	lock        sync.RWMutex
+	requestID   string
+	dataStream  httpstream.Stream
+	errorStream httpstream.Stream
+	complete    chan struct{}
+}
+
+// newPortForwardPair creates a new httpStreamPair.
+func newPortForwardPair(requestID string) *httpStreamPair {
+	return &httpStreamPair{
+		requestID: requestID,
+		complete:  make(chan struct{}),
+	}
+}
+
+// add adds the stream to the httpStreamPair. If the pair already
+// contains a stream for the new stream's type, an error is returned. add
+// returns true if both the data and error streams for this pair have been
+// received.
+func (p *httpStreamPair) add(stream httpstream.Stream) (bool, error) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	switch stream.Headers().Get(api.StreamType) {
+	case api.StreamTypeError:
+		if p.errorStream != nil {
+			return false, errors.New("error stream already assigned")
+		}
+		p.errorStream = stream
+	case api.StreamTypeData:
+		if p.dataStream != nil {
+			return false, errors.New("data stream already assigned")
+		}
+		p.dataStream = stream
+	}
+
+	complete := p.errorStream != nil && p.dataStream != nil
+	if complete {
+		close(p.complete)
+	}
+	return complete, nil
+}
+
+// printError writes s to p.errorStream if p.errorStream has been set.
+func (p *httpStreamPair) printError(s string) {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+	if p.errorStream != nil {
+		fmt.Fprint(p.errorStream, s)
+	}
+}

internal/kubernetes/portforward/httpstream_test.go

@@ -0,0 +1,267 @@
+/*
+Copyright 2016 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 portforward
+
+import (
+	"net/http"
+	"testing"
+	"time"
+
+	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/httpstream"
+)
+
+func TestHTTPStreamReceived(t *testing.T) {
+	tests := map[string]struct {
+		port          string
+		streamType    string
+		expectedError string
+	}{
+		"missing port": {
+			expectedError: `"port" header is required`,
+		},
+		"unable to parse port": {
+			port:          "abc",
+			expectedError: `unable to parse "abc" as a port: strconv.ParseUint: parsing "abc": invalid syntax`,
+		},
+		"negative port": {
+			port:          "-1",
+			expectedError: `unable to parse "-1" as a port: strconv.ParseUint: parsing "-1": invalid syntax`,
+		},
+		"missing stream type": {
+			port:          "80",
+			expectedError: `"streamType" header is required`,
+		},
+		"valid port with error stream": {
+			port:       "80",
+			streamType: "error",
+		},
+		"valid port with data stream": {
+			port:       "80",
+			streamType: "data",
+		},
+		"invalid stream type": {
+			port:          "80",
+			streamType:    "foo",
+			expectedError: `invalid stream type "foo"`,
+		},
+	}
+	for name, test := range tests {
+		streams := make(chan httpstream.Stream, 1)
+		f := httpStreamReceived(streams)
+		stream := newFakeHTTPStream()
+		if len(test.port) > 0 {
+			stream.headers.Set("port", test.port)
+		}
+		if len(test.streamType) > 0 {
+			stream.headers.Set("streamType", test.streamType)
+		}
+		replySent := make(chan struct{})
+		err := f(stream, replySent)
+		close(replySent)
+		if len(test.expectedError) > 0 {
+			if err == nil {
+				t.Errorf("%s: expected err=%q, but it was nil", name, test.expectedError)
+			}
+			if e, a := test.expectedError, err.Error(); e != a {
+				t.Errorf("%s: expected err=%q, got %q", name, e, a)
+			}
+			continue
+		}
+		if err != nil {
+			t.Errorf("%s: unexpected error %v", name, err)
+			continue
+		}
+		if s := <-streams; s != stream {
+			t.Errorf("%s: expected stream %#v, got %#v", name, stream, s)
+		}
+	}
+}
+
+type fakeConn struct {
+	removeStreamsCalled bool
+}
+
+func (*fakeConn) CreateStream(headers http.Header) (httpstream.Stream, error) { return nil, nil }
+func (*fakeConn) Close() error                                                { return nil }
+func (*fakeConn) CloseChan() <-chan bool                                      { return nil }
+func (*fakeConn) SetIdleTimeout(timeout time.Duration)                        {}
+func (f *fakeConn) RemoveStreams(streams ...httpstream.Stream)                { f.removeStreamsCalled = true }
+
+func TestGetStreamPair(t *testing.T) {
+	timeout := make(chan time.Time)
+
+	conn := &fakeConn{}
+	h := &httpStreamHandler{
+		streamPairs: make(map[string]*httpStreamPair),
+		conn:        conn,
+	}
+
+	// test adding a new entry
+	p, created := h.getStreamPair("1")
+	if p == nil {
+		t.Fatalf("unexpected nil pair")
+	}
+	if !created {
+		t.Fatal("expected created=true")
+	}
+	if p.dataStream != nil {
+		t.Errorf("unexpected non-nil data stream")
+	}
+	if p.errorStream != nil {
+		t.Errorf("unexpected non-nil error stream")
+	}
+
+	// start the monitor for this pair
+	monitorDone := make(chan struct{})
+	go func() {
+		h.monitorStreamPair(p, timeout)
+		close(monitorDone)
+	}()
+
+	if !h.hasStreamPair("1") {
+		t.Fatal("This should still be true")
+	}
+
+	// make sure we can retrieve an existing entry
+	p2, created := h.getStreamPair("1")
+	if created {
+		t.Fatal("expected created=false")
+	}
+	if p != p2 {
+		t.Fatalf("retrieving an existing pair: expected %#v, got %#v", p, p2)
+	}
+
+	// removed via complete
+	dataStream := newFakeHTTPStream()
+	dataStream.headers.Set(api.StreamType, api.StreamTypeData)
+	complete, err := p.add(dataStream)
+	if err != nil {
+		t.Fatalf("unexpected error adding data stream to pair: %v", err)
+	}
+	if complete {
+		t.Fatalf("unexpected complete")
+	}
+
+	errorStream := newFakeHTTPStream()
+	errorStream.headers.Set(api.StreamType, api.StreamTypeError)
+	complete, err = p.add(errorStream)
+	if err != nil {
+		t.Fatalf("unexpected error adding error stream to pair: %v", err)
+	}
+	if !complete {
+		t.Fatal("unexpected incomplete")
+	}
+
+	// make sure monitorStreamPair completed
+	<-monitorDone
+
+	if !conn.removeStreamsCalled {
+		t.Fatalf("connection remove stream not called")
+	}
+	conn.removeStreamsCalled = false
+
+	// make sure the pair was removed
+	if h.hasStreamPair("1") {
+		t.Fatal("expected removal of pair after both data and error streams received")
+	}
+
+	// removed via timeout
+	p, created = h.getStreamPair("2")
+	if !created {
+		t.Fatal("expected created=true")
+	}
+	if p == nil {
+		t.Fatal("expected p not to be nil")
+	}
+
+	monitorDone = make(chan struct{})
+	go func() {
+		h.monitorStreamPair(p, timeout)
+		close(monitorDone)
+	}()
+	// cause the timeout
+	close(timeout)
+	// make sure monitorStreamPair completed
+	<-monitorDone
+	if h.hasStreamPair("2") {
+		t.Fatal("expected stream pair to be removed")
+	}
+	if !conn.removeStreamsCalled {
+		t.Fatalf("connection remove stream not called")
+	}
+}
+
+func TestRequestID(t *testing.T) {
+	h := &httpStreamHandler{}
+
+	s := newFakeHTTPStream()
+	s.headers.Set(api.StreamType, api.StreamTypeError)
+	s.id = 1
+	if e, a := "1", h.requestID(s); e != a {
+		t.Errorf("expected %q, got %q", e, a)
+	}
+
+	s.headers.Set(api.StreamType, api.StreamTypeData)
+	s.id = 3
+	if e, a := "1", h.requestID(s); e != a {
+		t.Errorf("expected %q, got %q", e, a)
+	}
+
+	s.id = 7
+	s.headers.Set(api.PortForwardRequestIDHeader, "2")
+	if e, a := "2", h.requestID(s); e != a {
+		t.Errorf("expected %q, got %q", e, a)
+	}
+}
+
+type fakeHTTPStream struct {
+	headers http.Header
+	id      uint32
+}
+
+func newFakeHTTPStream() *fakeHTTPStream {
+	return &fakeHTTPStream{
+		headers: make(http.Header),
+	}
+}
+
+var _ httpstream.Stream = &fakeHTTPStream{}
+
+func (s *fakeHTTPStream) Read(data []byte) (int, error) {
+	return 0, nil
+}
+
+func (s *fakeHTTPStream) Write(data []byte) (int, error) {
+	return 0, nil
+}
+
+func (s *fakeHTTPStream) Close() error {
+	return nil
+}
+
+func (s *fakeHTTPStream) Reset() error {
+	return nil
+}
+
+func (s *fakeHTTPStream) Headers() http.Header {
+	return s.headers
+}
+
+func (s *fakeHTTPStream) Identifier() uint32 {
+	return s.id
+}

internal/kubernetes/portforward/portforward.go

@@ -0,0 +1,54 @@
+/*
+Copyright 2016 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 portforward
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apiserver/pkg/util/wsstream"
+)
+
+// PortForwarder knows how to forward content from a data stream to/from a port
+// in a pod.
+type PortForwarder interface {
+	// PortForwarder copies data between a data stream and a port in a pod.
+	PortForward(ctx context.Context, name string, uid types.UID, port int32, stream io.ReadWriteCloser) error
+}
+
+// ServePortForward handles a port forwarding request.  A single request is
+// kept alive as long as the client is still alive and the connection has not
+// been timed out due to idleness. This function handles multiple forwarded
+// connections; i.e., multiple `curl http://localhost:8888/` requests will be
+// handled by a single invocation of ServePortForward.
+func ServePortForward(w http.ResponseWriter, req *http.Request, portForwarder PortForwarder, podName string, uid types.UID, portForwardOptions *V4Options, idleTimeout time.Duration, streamCreationTimeout time.Duration, supportedProtocols []string) {
+	var err error
+	if wsstream.IsWebSocketRequest(req) {
+		err = handleWebSocketStreams(req, w, portForwarder, podName, uid, portForwardOptions, supportedProtocols, idleTimeout, streamCreationTimeout)
+	} else {
+		err = handleHTTPStreams(req, w, portForwarder, podName, uid, supportedProtocols, idleTimeout, streamCreationTimeout)
+	}
+
+	if err != nil {
+		runtime.HandleError(err)
+		return
+	}
+}

internal/kubernetes/portforward/websocket.go

@@ -0,0 +1,199 @@
+/*
+Copyright 2016 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 portforward
+
+import (
+	"context"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"k8s.io/klog/v2"
+
+	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apiserver/pkg/endpoints/responsewriter"
+	"k8s.io/apiserver/pkg/util/wsstream"
+)
+
+const (
+	dataChannel = iota
+	errorChannel
+
+	v4BinaryWebsocketProtocol = "v4." + wsstream.ChannelWebSocketProtocol
+	v4Base64WebsocketProtocol = "v4." + wsstream.Base64ChannelWebSocketProtocol
+)
+
+// V4Options contains details about which streams are required for port
+// forwarding.
+// All fields included in V4Options need to be expressed explicitly in the
+// CRI (k8s.io/cri-api/pkg/apis/{version}/api.proto) PortForwardRequest.
+type V4Options struct {
+	Ports []int32
+}
+
+// NewV4Options creates a new options from the Request.
+func NewV4Options(req *http.Request) (*V4Options, error) {
+	if !wsstream.IsWebSocketRequest(req) {
+		return &V4Options{}, nil
+	}
+
+	portStrings := req.URL.Query()[api.PortHeader]
+	if len(portStrings) == 0 {
+		return nil, fmt.Errorf("query parameter %q is required", api.PortHeader)
+	}
+
+	ports := make([]int32, 0, len(portStrings))
+	for _, portString := range portStrings {
+		if len(portString) == 0 {
+			return nil, fmt.Errorf("query parameter %q cannot be empty", api.PortHeader)
+		}
+		for _, p := range strings.Split(portString, ",") {
+			port, err := strconv.ParseUint(p, 10, 16)
+			if err != nil {
+				return nil, fmt.Errorf("unable to parse %q as a port: %v", portString, err)
+			}
+			if port < 1 {
+				return nil, fmt.Errorf("port %q must be > 0", portString)
+			}
+			ports = append(ports, int32(port))
+		}
+	}
+
+	return &V4Options{
+		Ports: ports,
+	}, nil
+}
+
+// BuildV4Options returns a V4Options based on the given information.
+func BuildV4Options(ports []int32) (*V4Options, error) {
+	return &V4Options{Ports: ports}, nil
+}
+
+// handleWebSocketStreams handles requests to forward ports to a pod via
+// a PortForwarder. A pair of streams are created per port (DATA n,
+// ERROR n+1). The associated port is written to each stream as a unsigned 16
+// bit integer in little endian format.
+func handleWebSocketStreams(req *http.Request, w http.ResponseWriter, portForwarder PortForwarder, podName string, uid types.UID, opts *V4Options, supportedPortForwardProtocols []string, idleTimeout, streamCreationTimeout time.Duration) error {
+	channels := make([]wsstream.ChannelType, 0, len(opts.Ports)*2)
+	for i := 0; i < len(opts.Ports); i++ {
+		channels = append(channels, wsstream.ReadWriteChannel, wsstream.WriteChannel)
+	}
+	conn := wsstream.NewConn(map[string]wsstream.ChannelProtocolConfig{
+		"": {
+			Binary:   true,
+			Channels: channels,
+		},
+		v4BinaryWebsocketProtocol: {
+			Binary:   true,
+			Channels: channels,
+		},
+		v4Base64WebsocketProtocol: {
+			Binary:   false,
+			Channels: channels,
+		},
+	})
+	conn.SetIdleTimeout(idleTimeout)
+	_, streams, err := conn.Open(responsewriter.GetOriginal(w), req)
+	if err != nil {
+		err = fmt.Errorf("unable to upgrade websocket connection: %v", err)
+		return err
+	}
+	defer conn.Close()
+	streamPairs := make([]*websocketStreamPair, len(opts.Ports))
+	for i := range streamPairs {
+		streamPair := websocketStreamPair{
+			port:        opts.Ports[i],
+			dataStream:  streams[i*2+dataChannel],
+			errorStream: streams[i*2+errorChannel],
+		}
+		streamPairs[i] = &streamPair
+
+		portBytes := make([]byte, 2)
+		// port is always positive so conversion is allowable
+		binary.LittleEndian.PutUint16(portBytes, uint16(streamPair.port))
+		streamPair.dataStream.Write(portBytes)
+		streamPair.errorStream.Write(portBytes)
+	}
+	h := &websocketStreamHandler{
+		conn:        conn,
+		streamPairs: streamPairs,
+		pod:         podName,
+		uid:         uid,
+		forwarder:   portForwarder,
+	}
+	h.run()
+
+	return nil
+}
+
+// websocketStreamPair represents the error and data streams for a port
+// forwarding request.
+type websocketStreamPair struct {
+	port        int32
+	dataStream  io.ReadWriteCloser
+	errorStream io.WriteCloser
+}
+
+// websocketStreamHandler is capable of processing a single port forward
+// request over a websocket connection
+type websocketStreamHandler struct {
+	conn        *wsstream.Conn
+	streamPairs []*websocketStreamPair
+	pod         string
+	uid         types.UID
+	forwarder   PortForwarder
+}
+
+// run invokes the websocketStreamHandler's forwarder.PortForward
+// function for the given stream pair.
+func (h *websocketStreamHandler) run() {
+	wg := sync.WaitGroup{}
+	wg.Add(len(h.streamPairs))
+
+	for _, pair := range h.streamPairs {
+		p := pair
+		go func() {
+			defer wg.Done()
+			h.portForward(p)
+		}()
+	}
+
+	wg.Wait()
+}
+
+func (h *websocketStreamHandler) portForward(p *websocketStreamPair) {
+	ctx := context.Background()
+	defer p.dataStream.Close()
+	defer p.errorStream.Close()
+
+	klog.V(5).InfoS("Connection invoking forwarder.PortForward for port", "connection", h.conn, "port", p.port)
+	err := h.forwarder.PortForward(ctx, h.pod, h.uid, p.port, p.dataStream)
+	klog.V(5).InfoS("Connection done invoking forwarder.PortForward for port", "connection", h.conn, "port", p.port)
+
+	if err != nil {
+		msg := fmt.Errorf("error forwarding port %d to pod %s, uid %v: %v", p.port, h.pod, h.uid, err)
+		runtime.HandleError(msg)
+		fmt.Fprint(p.errorStream, msg.Error())
+	}
+}

internal/kubernetes/portforward/websocket_test.go

@@ -0,0 +1,101 @@
+/*
+Copyright 2016 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 portforward
+
+import (
+	"net/http"
+	"reflect"
+	"testing"
+)
+
+func TestV4Options(t *testing.T) {
+	tests := map[string]struct {
+		url           string
+		websocket     bool
+		expectedOpts  *V4Options
+		expectedError string
+	}{
+		"non-ws request": {
+			url:          "http://example.com",
+			expectedOpts: &V4Options{},
+		},
+		"missing port": {
+			url:           "http://example.com",
+			websocket:     true,
+			expectedError: `query parameter "port" is required`,
+		},
+		"unable to parse port": {
+			url:           "http://example.com?port=abc",
+			websocket:     true,
+			expectedError: `unable to parse "abc" as a port: strconv.ParseUint: parsing "abc": invalid syntax`,
+		},
+		"negative port": {
+			url:           "http://example.com?port=-1",
+			websocket:     true,
+			expectedError: `unable to parse "-1" as a port: strconv.ParseUint: parsing "-1": invalid syntax`,
+		},
+		"one port": {
+			url:       "http://example.com?port=80",
+			websocket: true,
+			expectedOpts: &V4Options{
+				Ports: []int32{80},
+			},
+		},
+		"multiple ports": {
+			url:       "http://example.com?port=80,90,100",
+			websocket: true,
+			expectedOpts: &V4Options{
+				Ports: []int32{80, 90, 100},
+			},
+		},
+		"multiple port": {
+			url:       "http://example.com?port=80&port=90",
+			websocket: true,
+			expectedOpts: &V4Options{
+				Ports: []int32{80, 90},
+			},
+		},
+	}
+	for name, test := range tests {
+		req, err := http.NewRequest(http.MethodGet, test.url, nil)
+		if err != nil {
+			t.Errorf("%s: invalid url %q err=%q", name, test.url, err)
+			continue
+		}
+		if test.websocket {
+			req.Header.Set("Connection", "Upgrade")
+			req.Header.Set("Upgrade", "websocket")
+		}
+		opts, err := NewV4Options(req)
+		if len(test.expectedError) > 0 {
+			if err == nil {
+				t.Errorf("%s: expected err=%q, but it was nil", name, test.expectedError)
+			}
+			if e, a := test.expectedError, err.Error(); e != a {
+				t.Errorf("%s: expected err=%q, got %q", name, e, a)
+			}
+			continue
+		}
+		if err != nil {
+			t.Errorf("%s: unexpected error %v", name, err)
+			continue
+		}
+		if !reflect.DeepEqual(test.expectedOpts, opts) {
+			t.Errorf("%s: expected options %#v, got %#v", name, test.expectedOpts, err)
+		}
+	}
+}

internal/kubernetes/remotecommand/attach.go

@@ -0,0 +1,79 @@
+/*
+Copyright 2016 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 remotecommand
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/remotecommand"
+	utilexec "k8s.io/utils/exec"
+)
+
+// Attacher knows how to attach to a container in a pod.
+type Attacher interface {
+	// AttachToContainer attaches to a container in the pod, copying data
+	// between in/out/err and the container's stdin/stdout/stderr.
+	AttachToContainer(name string, uid types.UID, container string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error
+}
+
+// ServeAttach handles requests to attach to a container. After
+// creating/receiving the required streams, it delegates the actual attachment
+// to the attacher.
+func ServeAttach(w http.ResponseWriter, req *http.Request, attacher Attacher, podName string, uid types.UID, container string, streamOpts *Options, idleTimeout, streamCreationTimeout time.Duration, supportedProtocols []string) {
+	ctx, ok := createStreams(req, w, streamOpts, supportedProtocols, idleTimeout, streamCreationTimeout)
+	if !ok {
+		// error is handled by createStreams
+		return
+	}
+	defer ctx.conn.Close()
+
+	err := attacher.AttachToContainer(podName, uid, container, ctx.stdinStream, ctx.stdoutStream, ctx.stderrStream, ctx.tty, ctx.resizeChan, 0)
+	if err != nil {
+		if exitErr, ok := err.(utilexec.ExitError); ok && exitErr.Exited() {
+			rc := exitErr.ExitStatus()
+			ctx.writeStatus(&apierrors.StatusError{ErrStatus: metav1.Status{
+				Status: metav1.StatusFailure,
+				Reason: remotecommandconsts.NonZeroExitCodeReason,
+				Details: &metav1.StatusDetails{
+					Causes: []metav1.StatusCause{
+						{
+							Type:    remotecommandconsts.ExitCodeCauseType,
+							Message: fmt.Sprintf("%d", rc),
+						},
+					},
+				},
+				Message: fmt.Sprintf("command terminated with non-zero exit code: %v", exitErr),
+			}})
+			return
+		}
+		err = fmt.Errorf("error attaching to container: %v", err)
+		runtime.HandleError(err)
+		ctx.writeStatus(apierrors.NewInternalError(err))
+		return
+	}
+	ctx.writeStatus(&apierrors.StatusError{ErrStatus: metav1.Status{
+		Status: metav1.StatusSuccess,
+	}})
+}

internal/kubernetes/remotecommand/httpstream.go

@@ -148,6 +148,8 @@ func createHTTPStreamStreams(req *http.Request, w http.ResponseWriter, opts *Opt
 
 	var handler protocolHandler
 	switch protocol {
+	case remotecommandconsts.StreamProtocolV5Name:
+		handler = &v5ProtocolHandler{}
 	case remotecommandconsts.StreamProtocolV4Name:
 		handler = &v4ProtocolHandler{}
 	case remotecommandconsts.StreamProtocolV3Name:
@@ -159,6 +161,9 @@ func createHTTPStreamStreams(req *http.Request, w http.ResponseWriter, opts *Opt
 		fallthrough
 	case remotecommandconsts.StreamProtocolV1Name:
 		handler = &v1ProtocolHandler{}
+	default:
+		klog.Errorf("unable to create HTTP stream: unknown protocol %q", protocol)
+		return nil, false
 	}
 
 	// count the streams client asked for, starting with 1
@@ -199,6 +204,57 @@ type protocolHandler interface {
 	supportsTerminalResizing() bool
 }
 
+// v5ProtocolHandler implements the V5 protocol version for streaming command execution.
+type v5ProtocolHandler struct{}
+
+func (*v5ProtocolHandler) waitForStreams(streams <-chan streamAndReply, expectedStreams int, expired <-chan time.Time) (*context, error) {
+	ctx := &context{}
+	receivedStreams := 0
+	replyChan := make(chan struct{})
+	stop := make(chan struct{})
+	defer close(stop)
+WaitForStreams:
+	for {
+		select {
+		case stream := <-streams:
+			streamType := stream.Headers().Get(api.StreamType)
+			switch streamType {
+			case api.StreamTypeError:
+				ctx.writeStatus = v4WriteStatusFunc(stream) // write json errors
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdin:
+				ctx.stdinStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdout:
+				ctx.stdoutStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStderr:
+				ctx.stderrStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeResize:
+				ctx.resizeStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			default:
+				runtime.HandleError(fmt.Errorf("unexpected stream type: %q", streamType))
+			}
+		case <-replyChan:
+			receivedStreams++
+			if receivedStreams == expectedStreams {
+				break WaitForStreams
+			}
+		case <-expired:
+			// TODO find a way to return the error to the user. Maybe use a separate
+			// stream to report errors?
+			return nil, errors.New("timed out waiting for client to create streams")
+		}
+	}
+
+	return ctx, nil
+}
+
+// supportsTerminalResizing returns true because v5ProtocolHandler supports it
+func (*v5ProtocolHandler) supportsTerminalResizing() bool { return true }
+
 // v4ProtocolHandler implements the V4 protocol version for streaming command execution. It only differs
 // in from v3 in the error stream format using an json-marshaled metav1.Status which carries
 // the process' exit code.

internal/podutils/env.go

@@ -29,7 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	apivalidation "k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/client-go/tools/record"
-	"k8s.io/utils/pointer"
+	"k8s.io/utils/ptr"
 )
 
 const (
@@ -334,7 +334,7 @@ func getEnvironmentVariableValue(ctx context.Context, env *corev1.EnvVar, mappin
 		return getEnvironmentVariableValueWithValueFrom(ctx, env, mappingFunc, pod, container, rm, recorder)
 	}
 	// Handle values that have been directly provided after expanding variable references.
-	return pointer.StringPtr(expansion.Expand(env.Value, mappingFunc)), nil
+	return ptr.To(expansion.Expand(env.Value, mappingFunc)), nil
 }
 
 func getEnvironmentVariableValueWithValueFrom(ctx context.Context, env *corev1.EnvVar, mappingFunc func(string) string, pod *corev1.Pod, container *corev1.Container, rm *manager.ResourceManager, recorder record.EventRecorder) (*string, error) {
@@ -411,7 +411,7 @@ func getEnvironmentVariableValueWithValueFromConfigMapKeyRef(ctx context.Context
 		return nil, fmt.Errorf("configmap %q doesn't contain the %q key required by pod %s", vf.Name, vf.Key, pod.Name)
 	}
 	// Populate the environment variable and continue on to the next reference.
-	return pointer.StringPtr(keyValue), nil
+	return ptr.To(keyValue), nil
 }
 
 func getEnvironmentVariableValueWithValueFromSecretKeyRef(ctx context.Context, env *corev1.EnvVar, mappingFunc func(string) string, pod *corev1.Pod, container *corev1.Container, rm *manager.ResourceManager, recorder record.EventRecorder) (*string, error) {
@@ -463,7 +463,7 @@ func getEnvironmentVariableValueWithValueFromSecretKeyRef(ctx context.Context, e
 		return nil, fmt.Errorf("secret %q doesn't contain the %q key required by pod %s", vf.Name, vf.Key, pod.Name)
 	}
 	// Populate the environment variable and continue on to the next reference.
-	return pointer.StringPtr(string(keyValue)), nil
+	return ptr.To(string(keyValue)), nil
 }
 
 // Handle population from a field (downward API).
@@ -476,7 +476,7 @@ func getEnvironmentVariableValueWithValueFromFieldRef(ctx context.Context, env *
 		return nil, err
 	}
 
-	return pointer.StringPtr(runtimeVal), nil
+	return ptr.To(runtimeVal), nil
 }
 
 // podFieldSelectorRuntimeValue returns the runtime value of the given

internal/podutils/helper.go

@@ -111,15 +111,15 @@ func ExtractFieldPathAsString(obj interface{}, fieldPath string) (string, error)
 
 // SplitMaybeSubscriptedPath checks whether the specified fieldPath is
 // subscripted, and
-//  - if yes, this function splits the fieldPath into path and subscript, and
-//    returns (path, subscript, true).
-//  - if no, this function returns (fieldPath, "", false).
+//   - if yes, this function splits the fieldPath into path and subscript, and
+//     returns (path, subscript, true).
+//   - if no, this function returns (fieldPath, "", false).
 //
 // Example inputs and outputs:
-//  - "metadata.annotations['myKey']" --> ("metadata.annotations", "myKey", true)
-//  - "metadata.annotations['a[b]c']" --> ("metadata.annotations", "a[b]c", true)
-//  - "metadata.labels['']"           --> ("metadata.labels", "", true)
-//  - "metadata.labels"               --> ("metadata.labels", "", false)
+//   - "metadata.annotations['myKey']" --> ("metadata.annotations", "myKey", true)
+//   - "metadata.annotations['a[b]c']" --> ("metadata.annotations", "a[b]c", true)
+//   - "metadata.labels[”]"           --> ("metadata.labels", "", true)
+//   - "metadata.labels"               --> ("metadata.labels", "", false)
 func SplitMaybeSubscriptedPath(fieldPath string) (string, string, bool) {
 	if !strings.HasSuffix(fieldPath, "']") {
 		return fieldPath, "", false

internal/test/e2e/framework/pod.go

@@ -6,7 +6,6 @@ import (
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
-	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -106,9 +105,9 @@ func (f *Framework) WaitUntilPodReady(namespace, name string) (*corev1.Pod, erro
 }
 
 // IsPodReady returns true if a pod is ready.
-func IsPodReady(pod *v1.Pod) bool {
+func IsPodReady(pod *corev1.Pod) bool {
 	for _, cond := range pod.Status.Conditions {
-		if cond.Type == v1.PodReady && cond.Status == v1.ConditionTrue {
+		if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
 			return true
 		}
 	}

internal/test/e2e/framework/stats.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 
+	api "github.com/virtual-kubelet/virtual-kubelet/node/api"
 	stats "github.com/virtual-kubelet/virtual-kubelet/node/api/statsv1alpha1"
 	"k8s.io/apimachinery/pkg/util/net"
 )
@@ -29,3 +30,21 @@ func (f *Framework) GetStatsSummary(ctx context.Context) (*stats.Summary, error)
 	}
 	return res, nil
 }
+
+// GetStatsSummary queries the /metrics/resource endpoint of the virtual-kubelet and returns the Summary object obtained as a response.
+func (f *Framework) GetMetricsResource(ctx context.Context) ([]byte, error) {
+	// Query the /stats/summary endpoint.
+	b, err := f.KubeClient.CoreV1().
+		RESTClient().
+		Get().
+		Namespace(f.Namespace).
+		Resource("pods").
+		SubResource("proxy").
+		Name(net.JoinSchemeNamePort("https", f.NodeName, "10250")).
+		Suffix(api.MetricsResourceRouteSuffix).DoRaw(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return b, nil
+}

internal/test/e2e/main_test.go

@@ -1,3 +1,4 @@
+//go:build e2e
 // +build e2e
 
 package e2e
@@ -48,7 +49,7 @@ func teardown() error {
 	return nil
 }
 
-//  Provider-specific shouldSkipTest function
+// Provider-specific shouldSkipTest function
 func shouldSkipTest(testName string) bool {
 	return false
 }

node/api/attach.go

@@ -0,0 +1,139 @@
+// Copyright © 2017 The virtual-kubelet 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 api
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/virtual-kubelet/virtual-kubelet/errdefs"
+	"github.com/virtual-kubelet/virtual-kubelet/internal/kubernetes/remotecommand"
+	"k8s.io/apimachinery/pkg/types"
+	remoteutils "k8s.io/client-go/tools/remotecommand"
+)
+
+// ContainerAttachHandlerFunc defines the handler function used for "execing" into a
+// container in a pod.
+type ContainerAttachHandlerFunc func(ctx context.Context, namespace, podName, containerName string, attach AttachIO) error
+
+// HandleContainerAttach makes an http handler func from a Provider which execs a command in a pod's container
+// Note that this handler currently depends on gorrilla/mux to get url parts as variables.
+// TODO(@cpuguy83): don't force gorilla/mux on consumers of this function
+func HandleContainerAttach(h ContainerAttachHandlerFunc, opts ...ContainerExecHandlerOption) http.HandlerFunc {
+	if h == nil {
+		return NotImplemented
+	}
+
+	var cfg ContainerExecHandlerConfig
+	for _, o := range opts {
+		o(&cfg)
+	}
+
+	if cfg.StreamIdleTimeout == 0 {
+		cfg.StreamIdleTimeout = 30 * time.Second
+	}
+	if cfg.StreamCreationTimeout == 0 {
+		cfg.StreamCreationTimeout = 30 * time.Second
+	}
+
+	return handleError(func(w http.ResponseWriter, req *http.Request) error {
+		vars := mux.Vars(req)
+
+		namespace := vars["namespace"]
+		pod := vars["pod"]
+		container := vars["container"]
+
+		supportedStreamProtocols := strings.Split(req.Header.Get("X-Stream-Protocol-Version"), ",")
+
+		streamOpts, err := getExecOptions(req)
+		if err != nil {
+			return errdefs.AsInvalidInput(err)
+		}
+
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+
+		attach := &containerAttachContext{ctx: ctx, h: h, pod: pod, namespace: namespace, container: container}
+		remotecommand.ServeAttach(
+			w,
+			req,
+			attach,
+			"",
+			"",
+			container,
+			streamOpts,
+			cfg.StreamIdleTimeout,
+			cfg.StreamCreationTimeout,
+			supportedStreamProtocols,
+		)
+
+		return nil
+	})
+}
+
+type containerAttachContext struct {
+	h                         ContainerAttachHandlerFunc
+	namespace, pod, container string
+	ctx                       context.Context
+}
+
+// AttachToContainer Implements remotecommand.Attacher
+// This is called by remotecommand.ServeAttach
+func (c *containerAttachContext) AttachToContainer(name string, uid types.UID, container string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remoteutils.TerminalSize, timeout time.Duration) error {
+
+	eio := &execIO{
+		tty:    tty,
+		stdin:  in,
+		stdout: out,
+		stderr: err,
+	}
+
+	if tty {
+		eio.chResize = make(chan TermSize)
+	}
+
+	ctx, cancel := context.WithCancel(c.ctx)
+	defer cancel()
+
+	if tty {
+		go func() {
+			send := func(s remoteutils.TerminalSize) bool {
+				select {
+				case eio.chResize <- TermSize{Width: s.Width, Height: s.Height}:
+					return false
+				case <-ctx.Done():
+					return true
+				}
+			}
+
+			for {
+				select {
+				case s := <-resize:
+					if send(s) {
+						return
+					}
+				case <-ctx.Done():
+					return
+				}
+			}
+		}()
+	}
+
+	return c.h(c.ctx, c.namespace, c.pod, c.container, eio)
+}

node/api/metrics.go

@@ -0,0 +1,67 @@
+// Copyright © 2017 The virtual-kubelet 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 api
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+
+	"github.com/pkg/errors"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/expfmt"
+)
+
+const (
+	PrometheusTextFormatContentType = "text/plain; version=0.0.4"
+)
+
+// PodMetricsResourceHandlerFunc defines the handler for getting pod metrics
+type PodMetricsResourceHandlerFunc func(context.Context) ([]*dto.MetricFamily, error)
+
+// HandlePodMetricsResource makes an HTTP handler for implementing the kubelet /metrics/resource endpoint
+func HandlePodMetricsResource(h PodMetricsResourceHandlerFunc) http.HandlerFunc {
+	if h == nil {
+		return NotImplemented
+	}
+	return handleError(func(w http.ResponseWriter, req *http.Request) error {
+		metrics, err := h(req.Context())
+		if err != nil {
+			if isCancelled(err) {
+				return err
+			}
+			return errors.Wrap(err, "error getting status from provider")
+		}
+
+		// Convert metrics to Prometheus text format.
+		var buffer bytes.Buffer
+		enc := expfmt.NewEncoder(&buffer, expfmt.FmtText)
+		for _, mf := range metrics {
+			if err := enc.Encode(mf); err != nil {
+				return errors.Wrap(err, "could not convert metrics to prometheus text format")
+			}
+		}
+
+		// Set the response content type to "text/plain; version=0.0.4".
+		w.Header().Set("Content-Type", PrometheusTextFormatContentType)
+
+		// Write the metrics in Prometheus text format to the response writer.
+		if _, err := w.Write(buffer.Bytes()); err != nil {
+			return errors.Wrap(err, "could not write to client")
+		}
+
+		return nil
+	})
+}

node/api/metrics_test.go

@@ -0,0 +1,149 @@
+// Copyright © 2017 The virtual-kubelet 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 api_test
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/pkg/errors"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/virtual-kubelet/virtual-kubelet/node/api"
+	"google.golang.org/protobuf/proto"
+)
+
+const (
+	prometheusContentType = "text/plain; version=0.0.4"
+)
+
+func TestHandlePodMetricsResource(t *testing.T) {
+	testCases := []struct {
+		name               string
+		handler            api.PodMetricsResourceHandlerFunc
+		expectedStatusCode int
+		expectedError      error
+	}{
+		{
+			name: "Valid PodMetricsResourceHandlerFunc",
+			handler: func(_ context.Context) ([]*dto.MetricFamily, error) {
+				// Create the expected metrics.
+				cpuUsageMetric := &dto.MetricFamily{
+					Name:   proto.String("container_cpu_usage_seconds_total"),
+					Help:   proto.String("[ALPHA] Cumulative cpu time consumed by the container in core-seconds"),
+					Type:   dto.MetricType_GAUGE.Enum(),
+					Metric: []*dto.Metric{},
+				}
+				memoryUsageMetric := &dto.MetricFamily{
+					Name:   proto.String("container_memory_working_set_bytes"),
+					Help:   proto.String("[ALPHA] Current working set of the container in bytes"),
+					Type:   dto.MetricType_GAUGE.Enum(),
+					Metric: []*dto.Metric{},
+				}
+
+				// Add the sample metrics to the metric families.
+				cpuUsageMetric.Metric = append(cpuUsageMetric.Metric, createSampleMetric(
+					map[string]string{
+						"container": "simple-hello-world-container",
+						"namespace": "k8se-apps",
+						"pod":       "test-pod--zruwatj-86454fdc54-2wwpw",
+					},
+					0.1104636, 1680536423102,
+				))
+				cpuUsageMetric.Metric = append(cpuUsageMetric.Metric, createSampleMetric(
+					map[string]string{
+						"container": "simple-hello-world-container",
+						"namespace": "k8se-apps",
+						"pod":       "test-pod--zruwatj-86454fdc54-4mzd4",
+					},
+					0.11322, 1680536423103,
+				))
+				memoryUsageMetric.Metric = append(memoryUsageMetric.Metric, createSampleMetric(
+					map[string]string{
+						"container": "simple-hello-world-container",
+						"namespace": "k8se-apps",
+						"pod":       "test-pod--zruwatj-86454fdc54-2wwpw",
+					},
+					2.3277568e+07, 1680536423102,
+				))
+				memoryUsageMetric.Metric = append(memoryUsageMetric.Metric, createSampleMetric(
+					map[string]string{
+						"container": "simple-hello-world-container",
+						"namespace": "k8se-apps",
+						"pod":       "test-pod--zruwatj-86454fdc54-4mzd4",
+					},
+					2.2450176e+07, 1680536423104,
+				))
+
+				return []*dto.MetricFamily{cpuUsageMetric, memoryUsageMetric}, nil
+			},
+			expectedStatusCode: http.StatusOK,
+		},
+		{
+			name:               "Nil PodMetricsResourceHandlerFunc",
+			handler:            nil,
+			expectedStatusCode: http.StatusNotImplemented,
+		},
+		{
+			name: "Error in PodMetricsResourceHandlerFunc",
+			handler: func(_ context.Context) ([]*dto.MetricFamily, error) {
+				return nil, errors.New("test error")
+			},
+			expectedStatusCode: http.StatusInternalServerError,
+			expectedError:      errors.New("error getting status from provider: test error"),
+		},
+		// Add more test cases as needed
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			h := api.HandlePodMetricsResource(tc.handler)
+			require.NotNil(t, h)
+
+			rr := httptest.NewRecorder()
+			req, err := http.NewRequest("GET", "/metrics/resource", nil)
+			require.NoError(t, err)
+
+			h.ServeHTTP(rr, req)
+
+			assert.Equal(t, tc.expectedStatusCode, rr.Code)
+
+			if tc.expectedError != nil {
+				bodyBytes, err := io.ReadAll(rr.Body)
+				require.NoError(t, err)
+				assert.Contains(t, string(bodyBytes), tc.expectedError.Error())
+			} else if tc.expectedStatusCode == http.StatusOK {
+				contentType := rr.Header().Get("Content-Type")
+				assert.Equal(t, prometheusContentType, contentType)
+			}
+		})
+	}
+}
+
+func createSampleMetric(labels map[string]string, value float64, timestamp int64) *dto.Metric {
+	labelPairs := []*dto.LabelPair{}
+	for k, v := range labels {
+		labelPairs = append(labelPairs, &dto.LabelPair{
+			Name:  proto.String(k),
+			Value: proto.String(v),
+		})
+	}
+
+	return &dto.Metric{Label: labelPairs, Gauge: &dto.Gauge{Value: &value}, TimestampMs: &timestamp}
+}

node/api/portforward.go

@@ -0,0 +1,116 @@
+// Copyright © 2017 The virtual-kubelet 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 api
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/virtual-kubelet/virtual-kubelet/internal/kubernetes/portforward"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// PortForwardHandlerFunc defines the handler function used to
+// portforward, passing through the original dataStream
+type PortForwardHandlerFunc func(ctx context.Context, namespace, pod string, port int32, stream io.ReadWriteCloser) error
+
+// PortForwardHandlerConfig is used to pass options to options to the container exec handler.
+type PortForwardHandlerConfig struct {
+	// StreamIdleTimeout is the maximum time a streaming connection
+	// can be idle before the connection is automatically closed.
+	StreamIdleTimeout time.Duration
+	// StreamCreationTimeout is the maximum time for streaming connection
+	StreamCreationTimeout time.Duration
+}
+
+// PortForwardHandlerOption configures a PortForwardHandlerConfig
+// It is used as functional options passed to `HandlePortForward`
+type PortForwardHandlerOption func(*PortForwardHandlerConfig)
+
+// WithPortForwardStreamIdleTimeout sets the idle timeout for a container port forward streaming
+func WithPortForwardStreamIdleTimeout(dur time.Duration) PortForwardHandlerOption {
+	return func(cfg *PortForwardHandlerConfig) {
+		cfg.StreamIdleTimeout = dur
+	}
+}
+
+// WithPortForwardCreationTimeout sets the creation timeout for a container exec stream
+func WithPortForwardCreationTimeout(dur time.Duration) PortForwardHandlerOption {
+	return func(cfg *PortForwardHandlerConfig) {
+		cfg.StreamCreationTimeout = dur
+	}
+}
+
+// HandlePortForward makes an http handler func from a Provider which forward ports to a container
+// Note that this handler currently depends on gorrilla/mux to get url parts as variables.
+func HandlePortForward(h PortForwardHandlerFunc, opts ...PortForwardHandlerOption) http.HandlerFunc {
+	if h == nil {
+		return NotImplemented
+	}
+
+	var cfg PortForwardHandlerConfig
+	for _, o := range opts {
+		o(&cfg)
+	}
+
+	if cfg.StreamIdleTimeout == 0 {
+		cfg.StreamIdleTimeout = 30 * time.Second
+	}
+	if cfg.StreamCreationTimeout == 0 {
+		cfg.StreamCreationTimeout = 30 * time.Second
+	}
+
+	return handleError(func(w http.ResponseWriter, req *http.Request) error {
+		vars := mux.Vars(req)
+
+		namespace := vars["namespace"]
+
+		pod := vars["pod"]
+
+		supportedStreamProtocols := strings.Split(req.Header.Get("X-Stream-Protocol-Version"), ",")
+
+		portfwd := &portForwardContext{h: h, pod: pod, namespace: namespace}
+		portforward.ServePortForward(
+			w,
+			req,
+			portfwd,
+			pod,
+			"",
+			&portforward.V4Options{}, // This is only used for websocket connection
+			cfg.StreamIdleTimeout,
+			cfg.StreamCreationTimeout,
+			supportedStreamProtocols,
+		)
+
+		return nil
+	})
+
+}
+
+type portForwardContext struct {
+	h         PortForwardHandlerFunc
+	pod       string
+	namespace string
+}
+
+// PortForward Implements portforward.Portforwarder
+// This is called by portforward.ServePortForward
+func (p *portForwardContext) PortForward(ctx context.Context, name string, uid types.UID, port int32, stream io.ReadWriteCloser) error {
+	return p.h(ctx, p.namespace, p.pod, port, stream)
+}

node/api/server.go

@@ -34,17 +34,22 @@ type ServeMux interface {
 }
 
 type PodHandlerConfig struct { //nolint:golint
-	RunInContainer   ContainerExecHandlerFunc
-	GetContainerLogs ContainerLogsHandlerFunc
+	RunInContainer    ContainerExecHandlerFunc
+	AttachToContainer ContainerAttachHandlerFunc
+	PortForward       PortForwardHandlerFunc
+	GetContainerLogs  ContainerLogsHandlerFunc
 	// GetPods is meant to enumerate the pods that the provider knows about
 	GetPods PodListerFunc
 	// GetPodsFromKubernetes is meant to enumerate the pods that the node is meant to be running
 	GetPodsFromKubernetes PodListerFunc
 	GetStatsSummary       PodStatsSummaryHandlerFunc
+	GetMetricsResource    PodMetricsResourceHandlerFunc
 	StreamIdleTimeout     time.Duration
 	StreamCreationTimeout time.Duration
 }
 
+const MetricsResourceRouteSuffix = "/metrics/resource"
+
 // PodHandler creates an http handler for interacting with pods/containers.
 func PodHandler(p PodHandlerConfig, debug bool) http.Handler {
 	r := mux.NewRouter()
@@ -54,7 +59,6 @@ func PodHandler(p PodHandlerConfig, debug bool) http.Handler {
 	if debug {
 		r.HandleFunc("/runningpods/", HandleRunningPods(p.GetPods)).Methods("GET")
 	}
-
 	r.HandleFunc("/pods", HandleRunningPods(p.GetPodsFromKubernetes)).Methods("GET")
 	r.HandleFunc("/containerLogs/{namespace}/{pod}/{container}", HandleContainerLogs(p.GetContainerLogs)).Methods("GET")
 	r.HandleFunc(
@@ -65,6 +69,22 @@ func PodHandler(p PodHandlerConfig, debug bool) http.Handler {
 			WithExecStreamIdleTimeout(p.StreamIdleTimeout),
 		),
 	).Methods("POST", "GET")
+	r.HandleFunc(
+		"/attach/{namespace}/{pod}/{container}",
+		HandleContainerAttach(
+			p.AttachToContainer,
+			WithExecStreamCreationTimeout(p.StreamCreationTimeout),
+			WithExecStreamIdleTimeout(p.StreamIdleTimeout),
+		),
+	).Methods("POST", "GET")
+	r.HandleFunc(
+		"/portForward/{namespace}/{pod}",
+		HandlePortForward(
+			p.PortForward,
+			WithPortForwardStreamIdleTimeout(p.StreamCreationTimeout),
+			WithPortForwardCreationTimeout(p.StreamIdleTimeout),
+		),
+	).Methods("POST", "GET")
 
 	if p.GetStatsSummary != nil {
 		f := HandlePodStatsSummary(p.GetStatsSummary)
@@ -72,6 +92,11 @@ func PodHandler(p PodHandlerConfig, debug bool) http.Handler {
 		r.HandleFunc("/stats/summary/", f).Methods("GET")
 	}
 
+	if p.GetMetricsResource != nil {
+		f := HandlePodMetricsResource(p.GetMetricsResource)
+		r.HandleFunc(MetricsResourceRouteSuffix, f).Methods("GET")
+		r.HandleFunc(MetricsResourceRouteSuffix+"/", f).Methods("GET")
+	}
 	r.NotFoundHandler = http.HandlerFunc(NotFound)
 	return r
 }
@@ -97,6 +122,26 @@ func PodStatsSummaryHandler(f PodStatsSummaryHandlerFunc) http.Handler {
 	return r
 }
 
+// PodMetricsResourceHandler creates an http handler for serving pod metrics.
+//
+// If the passed in handler func is nil this will create handlers which only
+// serves http.StatusNotImplemented
+func PodMetricsResourceHandler(f PodMetricsResourceHandlerFunc) http.Handler {
+	if f == nil {
+		return http.HandlerFunc(NotImplemented)
+	}
+
+	r := mux.NewRouter()
+
+	h := HandlePodMetricsResource(f)
+
+	r.Handle(MetricsResourceRouteSuffix, ochttp.WithRouteTag(h, "PodMetricsResourceHandler")).Methods("GET")
+	r.Handle(MetricsResourceRouteSuffix+"/", ochttp.WithRouteTag(h, "PodMetricsResourceHandler")).Methods("GET")
+
+	r.NotFoundHandler = http.HandlerFunc(NotFound)
+	return r
+}
+
 // AttachPodRoutes adds the http routes for pod stuff to the passed in serve mux.
 //
 // Callers should take care to namespace the serve mux as they see fit, however
@@ -111,7 +156,8 @@ func AttachPodRoutes(p PodHandlerConfig, mux ServeMux, debug bool) {
 // The main reason for this struct is in case of expansion we do not need to break
 // the package level API.
 type PodMetricsConfig struct {
-	GetStatsSummary PodStatsSummaryHandlerFunc
+	GetStatsSummary    PodStatsSummaryHandlerFunc
+	GetMetricsResource PodMetricsResourceHandlerFunc
 }
 
 // AttachPodMetricsRoutes adds the http routes for pod/node metrics to the passed in serve mux.
@@ -120,6 +166,7 @@ type PodMetricsConfig struct {
 // these routes get called by the Kubernetes API server.
 func AttachPodMetricsRoutes(p PodMetricsConfig, mux ServeMux) {
 	mux.Handle("/", InstrumentHandler(HandlePodStatsSummary(p.GetStatsSummary)))
+	mux.Handle("/", InstrumentHandler(HandlePodMetricsResource(p.GetMetricsResource)))
 }
 
 func instrumentRequest(r *http.Request) *http.Request {

node/lease_controller_v1.go

@@ -30,7 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	coordclientset "k8s.io/client-go/kubernetes/typed/coordination/v1"
 	"k8s.io/utils/clock"
-	"k8s.io/utils/pointer"
+	"k8s.io/utils/ptr"
 )
 
 // Code heavily borrowed from: https://github.com/kubernetes/kubernetes/blob/v1.18.13/pkg/kubelet/nodelease/controller.go
@@ -275,8 +275,8 @@ func (c *leaseController) newLease(ctx context.Context, node *corev1.Node, base
 				Namespace: corev1.NamespaceNodeLease,
 			},
 			Spec: coordinationv1.LeaseSpec{
-				HolderIdentity:       pointer.StringPtr(node.Name),
-				LeaseDurationSeconds: pointer.Int32Ptr(c.leaseDurationSeconds),
+				HolderIdentity:       ptr.To(node.Name),
+				LeaseDurationSeconds: ptr.To(c.leaseDurationSeconds),
 			},
 		}
 	} else {

node/node_test.go

@@ -25,7 +25,6 @@ import (
 
 	"gotest.tools/assert"
 	"gotest.tools/assert/cmp"
-	is "gotest.tools/assert/cmp"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -394,8 +393,8 @@ func TestBeforeAnnotationsPreserved(t *testing.T) {
 	newNode, err := nodes.Get(ctx, testNodeCopy.Name, emptyGetOptions)
 	assert.NilError(t, err)
 
-	assert.Assert(t, is.Contains(newNode.Annotations, "testAnnotation"))
-	assert.Assert(t, is.Contains(newNode.Annotations, "beforeAnnotation"))
+	assert.Assert(t, cmp.Contains(newNode.Annotations, "testAnnotation"))
+	assert.Assert(t, cmp.Contains(newNode.Annotations, "beforeAnnotation"))
 }
 
 // Are conditions set by systems outside of VK preserved?
@@ -444,7 +443,7 @@ func TestManualConditionsPreserved(t *testing.T) {
 
 	newNode, err := nodes.Get(ctx, testNodeCopy.Name, emptyGetOptions)
 	assert.NilError(t, err)
-	assert.Assert(t, is.Len(newNode.Status.Conditions, 0))
+	assert.Assert(t, cmp.Len(newNode.Status.Conditions, 0))
 
 	baseCondition := corev1.NodeCondition{
 		Type:    "BaseCondition",
@@ -479,8 +478,8 @@ func TestManualConditionsPreserved(t *testing.T) {
 
 	newNode, err = nodes.Get(ctx, testNodeCopy.Name, emptyGetOptions)
 	assert.NilError(t, err)
-	assert.Assert(t, is.Len(newNode.Status.Conditions, 1))
-	assert.Assert(t, is.Contains(newNode.Annotations, "testAnnotation"))
+	assert.Assert(t, cmp.Len(newNode.Status.Conditions, 1))
+	assert.Assert(t, cmp.Contains(newNode.Annotations, "testAnnotation"))
 
 	// Add a new event manually
 	manuallyAddedCondition := corev1.NodeCondition{
@@ -507,8 +506,8 @@ func TestManualConditionsPreserved(t *testing.T) {
 				return true
 			}
 		}
-		assert.Assert(t, is.Contains(receivedNode.Annotations, "testAnnotation"))
-		assert.Assert(t, is.Contains(newNode.Annotations, "manuallyAddedAnnotation"))
+		assert.Assert(t, cmp.Contains(receivedNode.Annotations, "testAnnotation"))
+		assert.Assert(t, cmp.Contains(newNode.Annotations, "manuallyAddedAnnotation"))
 
 		return false
 	}))
@@ -553,11 +552,11 @@ func TestManualConditionsPreserved(t *testing.T) {
 	for idx := range newNode.Status.Conditions {
 		seenConditionTypes[idx] = newNode.Status.Conditions[idx].Type
 	}
-	assert.Assert(t, is.Contains(seenConditionTypes, baseCondition.Type))
-	assert.Assert(t, is.Contains(seenConditionTypes, newCondition.Type))
-	assert.Assert(t, is.Contains(seenConditionTypes, manuallyAddedCondition.Type))
-	assert.Assert(t, is.Equal(newNode.Annotations["testAnnotation"], ""))
-	assert.Assert(t, is.Contains(newNode.Annotations, "manuallyAddedAnnotation"))
+	assert.Assert(t, cmp.Contains(seenConditionTypes, baseCondition.Type))
+	assert.Assert(t, cmp.Contains(seenConditionTypes, newCondition.Type))
+	assert.Assert(t, cmp.Contains(seenConditionTypes, manuallyAddedCondition.Type))
+	assert.Assert(t, cmp.Equal(newNode.Annotations["testAnnotation"], ""))
+	assert.Assert(t, cmp.Contains(newNode.Annotations, "manuallyAddedAnnotation"))
 
 	t.Log(newNode.Status.Conditions)
 }
@@ -607,15 +606,15 @@ func TestNodePingSingleInflight(t *testing.T) {
 	assert.Assert(t, timeTakenToCompleteFirstPing < pingTimeout*5, "Time taken to complete first ping: %v", timeTakenToCompleteFirstPing)
 
 	assert.Assert(t, cmp.Error(firstPing.error, context.DeadlineExceeded.Error()))
-	assert.Assert(t, is.Equal(1, calls.read()))
-	assert.Assert(t, is.Equal(0, finished.read()))
+	assert.Assert(t, cmp.Equal(1, calls.read()))
+	assert.Assert(t, cmp.Equal(0, finished.read()))
 
 	// Wait until the first sleep finishes (the test context is done)
 	assert.NilError(t, finished.until(testCtx, func(i int) bool { return i > 0 }))
 
 	// Assert we didn't stack up goroutines, and that the one goroutine in flight finishd
-	assert.Assert(t, is.Equal(1, calls.read()))
-	assert.Assert(t, is.Equal(1, finished.read()))
+	assert.Assert(t, cmp.Equal(1, calls.read()))
+	assert.Assert(t, cmp.Equal(1, finished.read()))
 
 }
 

node/nodeutil/auth.go

@@ -14,6 +14,7 @@ import (
 	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
 	"k8s.io/apiserver/pkg/authorization/authorizerfactory"
+	"k8s.io/apiserver/pkg/server/options"
 	"k8s.io/client-go/kubernetes"
 )
 
@@ -114,13 +115,13 @@ type WebhookAuthConfig struct {
 func WebhookAuth(client kubernetes.Interface, nodeName string, opts ...WebhookAuthOption) (Auth, error) {
 	cfg := WebhookAuthConfig{
 		AuthnConfig: authenticatorfactory.DelegatingAuthenticatorConfig{
-			CacheTTL: 2 * time.Minute, // default taken from k8s.io/kubernetes/pkg/kubelet/apis/config/v1beta1
-			// TODO: After upgrading k8s libs, we need to add the retry backoff option
+			CacheTTL:            2 * time.Minute, // default taken from k8s.io/kubernetes/pkg/kubelet/apis/config/v1beta1
+			WebhookRetryBackoff: options.DefaultAuthWebhookRetryBackoff(),
 		},
 		AuthzConfig: authorizerfactory.DelegatingAuthorizerConfig{
-			AllowCacheTTL: 5 * time.Minute,  // default taken from k8s.io/kubernetes/pkg/kubelet/apis/config/v1beta1
-			DenyCacheTTL:  30 * time.Second, // default taken from k8s.io/kubernetes/pkg/kubelet/apis/config/v1beta1
-			// TODO: After upgrading k8s libs, we need to add the retry backoff option
+			AllowCacheTTL:       5 * time.Minute,  // default taken from k8s.io/kubernetes/pkg/kubelet/apis/config/v1beta1
+			DenyCacheTTL:        30 * time.Second, // default taken from k8s.io/kubernetes/pkg/kubelet/apis/config/v1beta1
+			WebhookRetryBackoff: options.DefaultAuthWebhookRetryBackoff(),
 		},
 	}
 

node/nodeutil/controller.go

@@ -251,6 +251,9 @@ type NodeConfig struct {
 	// The default value is derived from the number of cores available.
 	NumWorkers int
 
+	// Set the error handler for node status update failures
+	NodeStatusUpdateErrorHandler node.ErrorHandler
+
 	routeAttacher func(Provider, NodeConfig, corev1listers.PodLister)
 }
 
@@ -365,11 +368,19 @@ func NewNode(name string, newProvider NewProviderFunc, opts ...NodeOpt) (*Node,
 		}
 	}
 
+	nodeControllerOpts := []node.NodeControllerOpt{
+		node.WithNodeEnableLeaseV1(NodeLeaseV1Client(cfg.Client), node.DefaultLeaseDuration),
+	}
+
+	if cfg.NodeStatusUpdateErrorHandler != nil {
+		nodeControllerOpts = append(nodeControllerOpts, node.WithNodeStatusUpdateErrorHandler(cfg.NodeStatusUpdateErrorHandler))
+	}
+
 	nc, err := node.NewNodeController(
 		np,
 		&cfg.NodeSpec,
 		cfg.Client.CoreV1().Nodes(),
-		node.WithNodeEnableLeaseV1(NodeLeaseV1Client(cfg.Client), node.DefaultLeaseDuration),
+		nodeControllerOpts...,
 	)
 	if err != nil {
 		return nil, errors.Wrap(err, "error creating node controller")

node/nodeutil/provider.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"io"
 
+	dto "github.com/prometheus/client_model/go"
 	"github.com/virtual-kubelet/virtual-kubelet/node"
 	"github.com/virtual-kubelet/virtual-kubelet/node/api"
 	"github.com/virtual-kubelet/virtual-kubelet/node/api/statsv1alpha1"
@@ -27,8 +28,18 @@ type Provider interface {
 	// between in/out/err and the container's stdin/stdout/stderr.
 	RunInContainer(ctx context.Context, namespace, podName, containerName string, cmd []string, attach api.AttachIO) error
 
+	// AttachToContainer attaches to the executing process of a container in the pod, copying data
+	// between in/out/err and the container's stdin/stdout/stderr.
+	AttachToContainer(ctx context.Context, namespace, podName, containerName string, attach api.AttachIO) error
+
 	// GetStatsSummary gets the stats for the node, including running pods
 	GetStatsSummary(context.Context) (*statsv1alpha1.Summary, error)
+
+	// GetMetricsResource gets the metrics for the node, including running pods
+	GetMetricsResource(context.Context) ([]*dto.MetricFamily, error)
+
+	// PortForward forwards a local port to a port on the pod
+	PortForward(ctx context.Context, namespace, pod string, port int32, stream io.ReadWriteCloser) error
 }
 
 // ProviderConfig holds objects created by NewNodeFromClient that a provider may need to bootstrap itself.
@@ -54,15 +65,18 @@ func AttachProviderRoutes(mux api.ServeMux) NodeOpt {
 	return func(cfg *NodeConfig) error {
 		cfg.routeAttacher = func(p Provider, cfg NodeConfig, pods corev1listers.PodLister) {
 			mux.Handle("/", api.PodHandler(api.PodHandlerConfig{
-				RunInContainer:   p.RunInContainer,
-				GetContainerLogs: p.GetContainerLogs,
-				GetPods:          p.GetPods,
+				RunInContainer:    p.RunInContainer,
+				AttachToContainer: p.AttachToContainer,
+				GetContainerLogs:  p.GetContainerLogs,
+				GetPods:           p.GetPods,
 				GetPodsFromKubernetes: func(context.Context) ([]*v1.Pod, error) {
 					return pods.List(labels.Everything())
 				},
 				GetStatsSummary:       p.GetStatsSummary,
+				GetMetricsResource:    p.GetMetricsResource,
 				StreamIdleTimeout:     cfg.StreamIdleTimeout,
 				StreamCreationTimeout: cfg.StreamCreationTimeout,
+				PortForward:           p.PortForward,
 			}, true))
 		}
 		return nil

node/pod.go

@@ -139,7 +139,7 @@ func deleteGraceTimeEqual(old, new *int64) bool {
 	return false
 }
 
-// podShouldEnqueue checks if two pods equal according according to podsEqual func and DeleteTimeStamp
+// podShouldEnqueue checks if two pods equal according to podsEqual func and DeleteTimeStamp
 func podShouldEnqueue(oldPod, newPod *corev1.Pod) bool {
 	if !podsEqual(oldPod, newPod) {
 		return true
@@ -278,7 +278,7 @@ func (pc *PodController) enqueuePodStatusUpdate(ctx context.Context, pod *corev1
 	ctx = span.WithField(ctx, "key", key)
 
 	var obj interface{}
-	err = wait.PollImmediateUntil(notificationRetryPeriod, func() (bool, error) {
+	err = wait.PollUntilContextCancel(ctx, notificationRetryPeriod, true, func(ctx context.Context) (bool, error) {
 		var ok bool
 		obj, ok = pc.knownPods.Load(key)
 		if ok {
@@ -304,7 +304,7 @@ func (pc *PodController) enqueuePodStatusUpdate(ctx context.Context, pod *corev1
 		// that we're in some kind of startup synchronization issue where the provider knows about a pod (as it performs
 		// recover, that we do not yet know about).
 		return false, nil
-	}, ctx.Done())
+	})
 
 	if err != nil {
 		if errors.IsNotFound(err) {

node/podcontroller.go

@@ -402,7 +402,10 @@ func (pc *PodController) Run(ctx context.Context, podSyncWorkers int) (retErr er
 		}
 	}
 
-	pc.podsInformer.Informer().AddEventHandler(eventHandler)
+	_, err := pc.podsInformer.Informer().AddEventHandler(eventHandler)
+	if err != nil {
+		log.G(ctx).Error(err)
+	}
 
 	// Perform a reconciliation step that deletes any dangling pods from the provider.
 	// This happens only when the virtual-kubelet is starting, and operates on a "best-effort" basis.
@@ -450,6 +453,21 @@ func (pc *PodController) Err() error {
 	return pc.err
 }
 
+// SyncPodsFromKubernetesQueueLen returns the length of the SyncPodsFromKubernetes queue
+func (pc *PodController) SyncPodsFromKubernetesQueueLen() int {
+	return pc.syncPodsFromKubernetes.Len()
+}
+
+// DeletePodsFromKubernetesQueueLen returns the length of the DeletePodsFromKubernetes queue
+func (pc *PodController) DeletePodsFromKubernetesQueueLen() int {
+	return pc.deletePodsFromKubernetes.Len()
+}
+
+// SyncPodStatusFromProviderQueueLen returns the length of the SyncPodStatusFromProvider queue
+func (pc *PodController) SyncPodStatusFromProviderQueueLen() int {
+	return pc.syncPodStatusFromProvider.Len()
+}
+
 // syncPodFromKubernetesHandler compares the actual state with the desired, and attempts to converge the two.
 func (pc *PodController) syncPodFromKubernetesHandler(ctx context.Context, key string) error {
 	ctx, span := trace.StartSpan(ctx, "syncPodFromKubernetesHandler")

test/e2e/basic.go

@@ -1,11 +1,13 @@
 package e2e
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"testing"
 	"time"
 
+	"github.com/prometheus/common/expfmt"
 	"github.com/virtual-kubelet/virtual-kubelet/internal/podutils"
 	stats "github.com/virtual-kubelet/virtual-kubelet/node/api/statsv1alpha1"
 	"gotest.tools/assert"
@@ -111,6 +113,72 @@ func (ts *EndToEndTestSuite) TestGetStatsSummary(t *testing.T) {
 	}
 }
 
+// TestGetMetricsResource creates a pod having two containers and queries the /metrics/resource endpoint of the virtual-kubelet.
+// It expects this endpoint to return stats for the current node, as well as for the aforementioned pod and each of its two containers.
+func (ts *EndToEndTestSuite) TestGetMetricsResource(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a pod with prefix "nginx-" having three containers.
+	pod, err := f.CreatePod(ctx, f.CreateDummyPodObjectWithPrefix(t.Name(), "nginx", "foo", "bar", "baz"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Delete the "nginx-0-X" pod after the test finishes.
+	defer func() {
+		if err := f.DeletePodImmediately(ctx, pod.Namespace, pod.Name); err != nil && !apierrors.IsNotFound(err) {
+			t.Error(err)
+		}
+	}()
+
+	// Wait for the "nginx-" pod to be reported as running and ready.
+	if _, err := f.WaitUntilPodReady(pod.Namespace, pod.Name); err != nil {
+		t.Fatal(err)
+	}
+
+	// Grab the stats from the provider.
+	metricsResourceResponse, err := f.GetMetricsResource(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// decode metrics response bytes to metric family
+	reader := bytes.NewReader(metricsResourceResponse)
+	parser := expfmt.TextParser{}
+	metricsFamilyMap, err := parser.TextToMetricFamilies(reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Make sure the "nginx-" pod exists in the metrics returned.
+	currentContainerStatsCount := 0
+	found := false
+	for metricName, metricFamily := range metricsFamilyMap {
+		if metricName == "pod_cpu_usage_seconds_total" {
+			for _, metric := range metricFamily.Metric {
+				if *metric.Label[1].Value == pod.Name {
+					found = true
+				}
+			}
+		}
+		if metricName == "container_cpu_usage_seconds_total" {
+			for _, metric := range metricFamily.Metric {
+				if *metric.Label[1].Value == pod.Name {
+					currentContainerStatsCount += 1
+				}
+			}
+		}
+	}
+	if !found {
+		t.Fatalf("Pod %s not found in metrics", pod.Name)
+	}
+
+	// Make sure that we've got stats for all the containers in the "nginx-" pod.
+	desiredContainerStatsCount := len(pod.Spec.Containers)
+	if currentContainerStatsCount != desiredContainerStatsCount {
+		t.Fatalf("expected stats for %d containers, got stats for %d containers", desiredContainerStatsCount, currentContainerStatsCount)
+	}
+}
+
 // TestPodLifecycleGracefulDelete creates a pod and verifies that the provider has been asked to create it.
 // Then, it deletes the pods and verifies that the provider has been asked to delete it.
 // These verifications are made using the /stats/summary endpoint of the virtual-kubelet, by checking for the presence or absence of the pods.

trace/opentelemetry/opentelemetry.go

@@ -257,7 +257,7 @@ func makeAttribute(key string, val interface{}) (attr attribute.KeyValue) {
 	// 	return attribute.BoolSlice(key, v)
 	case error:
 		if v == nil {
-			attribute.String(key, "")
+			return attribute.String(key, "")
 		}
 		return attribute.String(key, v.Error())
 	default:

trace/opentelemetry/opentelemetry_test.go

@@ -30,8 +30,7 @@ import (
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
-	semconv "go.opentelemetry.io/otel/semconv"
-	"go.opentelemetry.io/otel/trace"
+	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
 	"gotest.tools/assert"
 	"gotest.tools/assert/cmp"
 )
@@ -92,7 +91,6 @@ func TestSetStatus(t *testing.T) {
 
 			assert.Assert(t, !s.s.IsRecording())
 			assert.Assert(t, e.status == tt.expectedCode)
-			assert.Assert(t, e.statusMessage == tt.expectedDescription)
 			s.SetStatus(tt.inputStatus) // should not be panic even if span is ended.
 		})
 	}
@@ -203,7 +201,7 @@ func TestLog(t *testing.T) {
 		logLevel           logLevel
 		fields             log.Fields
 		msg                string
-		expectedEvents     []trace.Event
+		expectedEvents     []sdktrace.Event
 		expectedAttributes []attribute.KeyValue
 	}{
 		{
@@ -212,7 +210,7 @@ func TestLog(t *testing.T) {
 			logLevel:           lDebug,
 			fields:             log.Fields{"testKey1": "value1"},
 			msg:                "message",
-			expectedEvents:     []trace.Event{{Name: "message"}},
+			expectedEvents:     []sdktrace.Event{{Name: "message"}},
 			expectedAttributes: []attribute.KeyValue{{Key: "testKey1", Value: attribute.StringValue("value1")}},
 		}, {
 			description:        "info",
@@ -220,7 +218,7 @@ func TestLog(t *testing.T) {
 			logLevel:           lInfo,
 			fields:             log.Fields{"testKey1": "value1"},
 			msg:                "message",
-			expectedEvents:     []trace.Event{{Name: "message"}},
+			expectedEvents:     []sdktrace.Event{{Name: "message"}},
 			expectedAttributes: []attribute.KeyValue{{Key: "testKey1", Value: attribute.StringValue("value1")}},
 		}, {
 			description:        "warn",
@@ -228,7 +226,7 @@ func TestLog(t *testing.T) {
 			logLevel:           lWarn,
 			fields:             log.Fields{"testKey1": "value1"},
 			msg:                "message",
-			expectedEvents:     []trace.Event{{Name: "message"}},
+			expectedEvents:     []sdktrace.Event{{Name: "message"}},
 			expectedAttributes: []attribute.KeyValue{{Key: "testKey1", Value: attribute.StringValue("value1")}},
 		}, {
 			description:        "error",
@@ -236,7 +234,7 @@ func TestLog(t *testing.T) {
 			logLevel:           lErr,
 			fields:             log.Fields{"testKey1": "value1"},
 			msg:                "message",
-			expectedEvents:     []trace.Event{{Name: "message"}},
+			expectedEvents:     []sdktrace.Event{{Name: "message"}},
 			expectedAttributes: []attribute.KeyValue{{Key: "testKey1", Value: attribute.StringValue("value1")}},
 		}, {
 			description:        "fatal",
@@ -244,7 +242,7 @@ func TestLog(t *testing.T) {
 			logLevel:           lFatal,
 			fields:             log.Fields{"testKey1": "value1"},
 			msg:                "message",
-			expectedEvents:     []trace.Event{{Name: "message"}},
+			expectedEvents:     []sdktrace.Event{{Name: "message"}},
 			expectedAttributes: []attribute.KeyValue{{Key: "testKey1", Value: attribute.StringValue("value1")}},
 		},
 	}
@@ -296,7 +294,7 @@ func TestLogf(t *testing.T) {
 		msg                string
 		fields             log.Fields
 		args               []interface{}
-		expectedEvents     []trace.Event
+		expectedEvents     []sdktrace.Event
 		expectedAttributes []attribute.KeyValue
 	}{
 		{
@@ -306,7 +304,7 @@ func TestLogf(t *testing.T) {
 			msg:            "k1: %s, k2: %v, k3: %d, k4: %v",
 			fields:         map[string]interface{}{"k1": "test", "k2": []string{"test"}, "k3": 1, "k4": []int{1}},
 			args:           []interface{}{"test", []string{"test"}, int(1), []int{1}},
-			expectedEvents: []trace.Event{{Name: "k1: test, k2: [test], k3: 1, k4: [1]"}},
+			expectedEvents: []sdktrace.Event{{Name: "k1: test, k2: [test], k3: 1, k4: [1]"}},
 			expectedAttributes: []attribute.KeyValue{
 				attribute.String("k1", "test"),
 				attribute.String("k2", fmt.Sprintf("%+v", []string{"test"})),
@@ -320,7 +318,7 @@ func TestLogf(t *testing.T) {
 			msg:            "k1: %d, k2: %v, k3: %f, k4: %v",
 			fields:         map[string]interface{}{"k1": int64(3), "k2": []int64{4}, "k3": float64(2), "k4": []float64{4}},
 			args:           []interface{}{int64(3), []int64{4}, float64(2), []float64{4}},
-			expectedEvents: []trace.Event{{Name: "k1: 3, k2: [4], k3: 2.000000, k4: [4]"}},
+			expectedEvents: []sdktrace.Event{{Name: "k1: 3, k2: [4], k3: 2.000000, k4: [4]"}},
 			expectedAttributes: []attribute.KeyValue{
 				attribute.Int64("k1", 1),
 				attribute.String("k2", fmt.Sprintf("%+v", []int64{2})),
@@ -334,7 +332,7 @@ func TestLogf(t *testing.T) {
 			msg:            "k1: %v, k2: %v",
 			fields:         map[string]interface{}{"k1": map[int]int{1: 1}, "k2": num(1)},
 			args:           []interface{}{map[int]int{1: 1}, num(1)},
-			expectedEvents: []trace.Event{{Name: "k1: map[1:1], k2: 1"}},
+			expectedEvents: []sdktrace.Event{{Name: "k1: map[1:1], k2: 1"}},
 			expectedAttributes: []attribute.KeyValue{
 				attribute.String("k1", "{1:1}"),
 				attribute.Stringer("k2", num(1)),
@@ -346,7 +344,7 @@ func TestLogf(t *testing.T) {
 			msg:            "k1: %t, k2: %v, k3: %s",
 			fields:         map[string]interface{}{"k1": true, "k2": []bool{true}, "k3": errors.New("fake")},
 			args:           []interface{}{true, []bool{true}, errors.New("fake")},
-			expectedEvents: []trace.Event{{Name: "k1: true, k2: [true], k3: fake"}},
+			expectedEvents: []sdktrace.Event{{Name: "k1: true, k2: [true], k3: fake"}},
 			expectedAttributes: []attribute.KeyValue{
 				attribute.Bool("k1", true),
 				attribute.String("k2", fmt.Sprintf("%+v", []bool{true})),
@@ -356,7 +354,7 @@ func TestLogf(t *testing.T) {
 			description:        "fatal",
 			spanName:           "test",
 			logLevel:           lFatal,
-			expectedEvents:     []trace.Event{{Name: ""}},
+			expectedEvents:     []sdktrace.Event{{Name: ""}},
 			expectedAttributes: []attribute.KeyValue{},
 		},
 	}
@@ -579,8 +577,7 @@ func setupSuite() (func(provider *sdktrace.TracerProvider), *sdktrace.TracerProv
 }
 
 func NewResource(name, version string) *resource.Resource {
-	return resource.NewWithAttributes(
-		semconv.ServiceNameKey.String(name),
+	return resource.NewWithAttributes(name,
 		semconv.ServiceVersionKey.String(version),
 	)
 }
@@ -590,36 +587,34 @@ type fakeExporter struct {
 	// attributes describe the aspects of the spans.
 	attributes []attribute.KeyValue
 	// Links returns all the links the span has to other spans.
-	links []trace.Link
+	links []sdktrace.Link
 	// Events returns all the events that occurred within in the spans
 	// lifetime.
-	events []trace.Event
+	events []sdktrace.Event
 	// Status returns the spans status.
-	status        codes.Code
-	statusMessage string
+	status codes.Code
 }
 
-func (f *fakeExporter) ExportSpans(_ context.Context, spans []*sdktrace.SpanSnapshot) error {
+func (f *fakeExporter) ExportSpans(_ context.Context, spans []sdktrace.ReadOnlySpan) error {
 	f.Lock()
 	defer f.Unlock()
 
 	f.attributes = make([]attribute.KeyValue, 0)
-	f.links = make([]trace.Link, 0)
-	f.events = make([]trace.Event, 0)
+	f.links = make([]sdktrace.Link, 0)
+	f.events = make([]sdktrace.Event, 0)
 	for _, s := range spans {
-		f.attributes = append(f.attributes, s.Attributes...)
-		f.links = append(f.links, s.Links...)
-		f.events = append(f.events, s.MessageEvents...)
-		f.status = s.StatusCode
-		f.statusMessage = s.StatusMessage
+		f.attributes = append(f.attributes, s.Attributes()...)
+		f.links = append(f.links, s.Links()...)
+		f.events = append(f.events, s.Events()...)
+		f.status = s.Status().Code
 	}
 	return nil
 }
 
 func (f *fakeExporter) Shutdown(_ context.Context) (err error) {
 	f.attributes = make([]attribute.KeyValue, 0)
-	f.links = make([]trace.Link, 0)
-	f.events = make([]trace.Event, 0)
+	f.links = make([]sdktrace.Link, 0)
+	f.events = make([]sdktrace.Event, 0)
 	return
 }
 

website/data/providers.yaml

@@ -26,3 +26,6 @@
   tag: openstack-zun
 - name: Tencent Games Tensile Kube
   tag: tensile-kube
+- name: StackPath Edge Compute
+  tag: virtual-kubelet-stackpath
+  org: stackpath