Wire up debug events

integration/debug_test.go

@@ -18,8 +18,10 @@ package integration
 
 import (
 	"testing"
+	"time"
 
 	"github.com/GoogleContainerTools/skaffold/integration/skaffold"
+	"github.com/GoogleContainerTools/skaffold/proto"
 )
 
 func TestDebug(t *testing.T) {
@@ -71,3 +73,67 @@ func TestDebug(t *testing.T) {
 		})
 	}
 }
+
+func TestDebugEventsRPC_StatusCheck(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	if RunOnGCP() {
+		t.Skip("skipping test that is not gcp only")
+	}
+
+	rpcAddr := randomPort()
+
+	// Run skaffold build first to fail quickly on a build failure
+	skaffold.Build().InDir("testdata/jib").RunOrFail(t)
+
+	ns, client, deleteNs := SetupNamespace(t)
+	defer deleteNs()
+
+	stop := skaffold.Debug("--enable-rpc", "--rpc-port", rpcAddr).InDir("testdata/jib").InNs(ns.Name).RunBackground(t)
+	defer stop()
+	waitForDebugEvent(t, client, rpcAddr)
+}
+
+func TestDebugEventsRPC_NoStatusCheck(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	if RunOnGCP() {
+		t.Skip("skipping test that is not gcp only")
+	}
+
+	rpcAddr := randomPort()
+
+	// Run skaffold build first to fail quickly on a build failure
+	skaffold.Build().InDir("testdata/jib").RunOrFail(t)
+
+	ns, client, deleteNs := SetupNamespace(t)
+	defer deleteNs()
+
+	stop := skaffold.Debug("--enable-rpc", "--rpc-port", rpcAddr, "--status-check=false").InDir("testdata/jib").InNs(ns.Name).RunBackground(t)
+	defer stop()
+	waitForDebugEvent(t, client, rpcAddr)
+}
+
+func waitForDebugEvent(t *testing.T, client *NSKubernetesClient, rpcAddr string) {
+	client.WaitForPodsReady()
+
+	_, entries, shutdown := apiEvents(t, rpcAddr)
+	defer shutdown()
+
+	timeout := time.After(1 * time.Minute)
+	for {
+		select {
+		case <-timeout:
+			t.Fatalf("timed out waiting for port debugging event")
+		case entry := <-entries:
+			switch entry.Event.GetEventType().(type) {
+			case *proto.Event_DebuggingContainerEvent:
+				// success!
+				return
+			default:
+			}
+		}
+	}
+}

pkg/skaffold/build/cache/cache.go

@@ -85,7 +85,7 @@ func NewCache(runCtx *runcontext.RunContext, imagesAreLocal bool, dependencies D
 		cacheFile:          cacheFile,
 		imagesAreLocal:     imagesAreLocal,
 		hashForArtifact: func(ctx context.Context, a *latest.Artifact) (string, error) {
-			return getHashForArtifact(ctx, dependencies, a, runCtx.DevMode)
+			return getHashForArtifact(ctx, dependencies, a, runCtx.IsDevMode())
 		},
 	}, nil
 }

pkg/skaffold/build/local/types.go

@@ -83,7 +83,7 @@ func NewBuilder(runCtx *runcontext.RunContext) (*Builder, error) {
 		localCluster:       localCluster,
 		pushImages:         pushImages,
 		skipTests:          runCtx.Opts.SkipTests,
-		devMode:            runCtx.DevMode,
+		devMode:            runCtx.IsDevMode(),
 		prune:              runCtx.Opts.Prune(),
 		pruneChildren:      !runCtx.Opts.NoPruneChildren,
 		insecureRegistries: runCtx.InsecureRegistries,

pkg/skaffold/debug/transform.go

@@ -114,8 +114,8 @@ const (
 	// debuggingSupportVolume is the name of the volume used to hold language runtime debugging support files.
 	debuggingSupportFilesVolume = "debugging-support-files"
 
-	// debugConfigAnnotation is the name of the podspec annotation that records debugging configuration information.
-	debugConfigAnnotation = "debug.cloud.google.com/config"
+	// DebugConfigAnnotation is the name of the podspec annotation that records debugging configuration information.
+	DebugConfigAnnotation = "debug.cloud.google.com/config"
 )
 
 var containerTransforms []containerTransformer
@@ -240,7 +240,7 @@ func transformPodSpec(metadata *metav1.ObjectMeta, podSpec *v1.PodSpec, retrieve
 		if metadata.Annotations == nil {
 			metadata.Annotations = make(map[string]string)
 		}
-		metadata.Annotations[debugConfigAnnotation] = encodeConfigurations(configurations)
+		metadata.Annotations[DebugConfigAnnotation] = encodeConfigurations(configurations)
 		return true
 	}
 	return false

pkg/skaffold/kubernetes/debugging/container_manager.go

@@ -0,0 +1,145 @@
+/*
+Copyright 2019 The Skaffold 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 debugging
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/watch"
+
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/debug"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/event"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
+)
+
+var (
+	// For testing
+	aggregatePodWatcher = kubernetes.AggregatePodWatcher
+
+	notifyDebuggingContainerStarted    = event.DebuggingContainerStarted
+	notifyDebuggingContainerTerminated = event.DebuggingContainerTerminated
+)
+
+type ContainerManager struct {
+	output      io.Writer
+	cli         *kubectl.CLI
+	podSelector kubernetes.PodSelector
+	namespaces  []string
+	active      map[string]string // set of containers that have been notified
+	aggregate   chan watch.Event
+}
+
+func NewContainerManager(out io.Writer, cli *kubectl.CLI, podSelector kubernetes.PodSelector, namespaces []string) *ContainerManager {
+	// Create the channel here as Stop() may be called before Start() when a build fails, thus
+	// avoiding the possibility of closing a nil channel. Channels are cheap.
+	return &ContainerManager{output: out, cli: cli, podSelector: podSelector, namespaces: namespaces, active: map[string]string{}, aggregate: make(chan watch.Event)}
+}
+
+func (d *ContainerManager) Start(ctx context.Context) error {
+	if d == nil {
+		// debug mode probably not enabled
+		return nil
+	}
+	stopWatchers, err := aggregatePodWatcher(d.namespaces, d.aggregate)
+	if err != nil {
+		stopWatchers()
+		return errors.Wrap(err, "initializing debugging container watcher")
+	}
+
+	go func() {
+		defer stopWatchers()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case evt, ok := <-d.aggregate:
+				if !ok {
+					return
+				}
+				// If the event's type is "ERROR", warn and continue.
+				if evt.Type == watch.Error {
+					logrus.Warnf("got unexpected event of type %s", evt.Type)
+					continue
+				}
+				// Grab the pod from the event.
+				pod, ok := evt.Object.(*v1.Pod)
+				if !ok {
+					continue
+				}
+				// Unlike other event watchers, we ignore event types as checkPod() uses only container status
+				if d.podSelector.Select(pod) {
+					d.checkPod(ctx, pod)
+				}
+			}
+		}
+	}()
+	return nil
+}
+
+func (d *ContainerManager) Stop() {
+	// if nil then debug mode probably not enabled
+	if d != nil {
+		close(d.aggregate)
+	}
+}
+
+func (d *ContainerManager) checkPod(_ context.Context, pod *v1.Pod) {
+	debugConfigString, found := pod.Annotations[debug.DebugConfigAnnotation]
+	if !found {
+		return
+	}
+	var configurations map[string]debug.ContainerDebugConfiguration
+	if err := json.Unmarshal([]byte(debugConfigString), &configurations); err != nil {
+		logrus.Warnf("Unable to parse debug-config for pod %s/%s: '%s'", pod.Namespace, pod.Name, debugConfigString)
+		return
+	}
+	for _, c := range pod.Status.ContainerStatuses {
+		// only examine debuggable containers
+		if config, found := configurations[c.Name]; found {
+			key := pod.Namespace + "/" + pod.Name + "/" + c.Name
+			// only notify of first appearance or disappearance
+			_, seen := d.active[key]
+			switch {
+			case c.State.Running != nil && !seen:
+				d.active[key] = key
+				notifyDebuggingContainerStarted(
+					pod.Name,
+					c.Name,
+					pod.Namespace,
+					config.Artifact,
+					config.Runtime,
+					config.WorkingDir,
+					config.Ports)
+
+			case c.State.Terminated != nil && seen:
+				delete(d.active, key)
+				notifyDebuggingContainerTerminated(pod.Name, c.Name, pod.Namespace,
+					config.Artifact,
+					config.Runtime,
+					config.WorkingDir,
+					config.Ports)
+			}
+		}
+	}
+}

pkg/skaffold/kubernetes/debugging/container_manager_test.go

@@ -0,0 +1,86 @@
+/*
+Copyright 2019 The Skaffold 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 debugging
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/GoogleContainerTools/skaffold/testutil"
+)
+
+func TestContainerManager(t *testing.T) {
+	testutil.Run(t, "simulation", func(t *testutil.T) {
+		startCount := 0
+		terminatedCount := 0
+		t.Override(&notifyDebuggingContainerStarted, func(podName string, containerName string, namespace string, artifactImage string, runtime string, workingDir string, debugPorts map[string]uint32) {
+			startCount++
+		})
+		t.Override(&notifyDebuggingContainerTerminated, func(podName string, containerName string, namespace string, artifactImage string, runtime string, workingDir string, debugPorts map[string]uint32) {
+			terminatedCount++
+		})
+		pod := v1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        "pod",
+				Namespace:   "ns",
+				Annotations: map[string]string{"debug.cloud.google.com/config": `{"test":{"runtime":"jvm","debugPorts":{"jdwp":5005}}}`},
+			},
+			Spec: v1.PodSpec{Containers: []v1.Container{
+				{
+					Name:    "test",
+					Command: []string{"java", "-jar", "foo.jar"},
+					Env:     []v1.EnvVar{{Name: "JAVA_TOOL_OPTIONS", Value: "-agentlib:jdwp=transport=dt_socket,server=y,address=5005,suspend=n,quiet=y"}},
+					Ports:   []v1.ContainerPort{{Name: "jdwp", ContainerPort: 5005}},
+				}}},
+			Status: v1.PodStatus{ContainerStatuses: []v1.ContainerStatus{{Name: "test", State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{}}}}},
+		}
+		m := &ContainerManager{output: &bytes.Buffer{}, active: make(map[string]string)}
+		state := &pod.Status.ContainerStatuses[0].State
+
+		// should never be active until running
+		m.checkPod(context.Background(), &pod)
+		t.CheckDeepEqual(0, len(m.active))
+		m.checkPod(context.Background(), &pod)
+		t.CheckDeepEqual(0, len(m.active))
+		t.CheckDeepEqual(0, startCount)
+		t.CheckDeepEqual(0, terminatedCount)
+
+		// container is now running
+		state.Waiting = nil
+		state.Running = &v1.ContainerStateRunning{}
+
+		m.checkPod(context.Background(), &pod)
+		t.CheckDeepEqual(1, len(m.active))
+		_, found := m.active["ns/pod/test"]
+		t.CheckDeepEqual(true, found)
+		t.CheckDeepEqual(1, startCount)
+		t.CheckDeepEqual(0, terminatedCount)
+
+		// container is now terminated
+		state.Running = nil
+		state.Terminated = &v1.ContainerStateTerminated{}
+
+		m.checkPod(context.Background(), &pod)
+		t.CheckDeepEqual(0, len(m.active))
+		t.CheckDeepEqual(1, startCount)
+		t.CheckDeepEqual(1, terminatedCount)
+	})
+}

pkg/skaffold/runner/debugging.go

@@ -0,0 +1,36 @@
+/*
+Copyright 2019 The Skaffold 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 runner
+
+import (
+	"io"
+
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/debugging"
+)
+
+func (r *SkaffoldRunner) createContainerManager(out io.Writer) {
+	if !r.runCtx.IsDebugMode() {
+		return
+	}
+	kubectlCLI := kubectl.NewFromRunContext(r.runCtx)
+	r.debugContainerManager = debugging.NewContainerManager(
+		out,
+		kubectlCLI,
+		r.podSelector,
+		r.runCtx.Namespaces)
+}