Merge pull request #1168 from dgageot/kaniko-builder

Improve Kaniko builder

integration/run_test.go

@@ -20,6 +20,7 @@ package integration
 
 import (
 	"bytes"
+	"context"
 	"flag"
 	"fmt"
 	"io/ioutil"
@@ -183,13 +184,13 @@ func TestRun(t *testing.T) {
 			}
 
 			for _, p := range testCase.pods {
-				if err := kubernetesutil.WaitForPodReady(client.CoreV1().Pods(ns.Name), p); err != nil {
+				if err := kubernetesutil.WaitForPodReady(context.Background(), client.CoreV1().Pods(ns.Name), p); err != nil {
 					t.Fatalf("Timed out waiting for pod ready")
 				}
 			}
 
 			for _, d := range testCase.deployments {
-				if err := kubernetesutil.WaitForDeploymentToStabilize(client, ns.Name, d, 10*time.Minute); err != nil {
+				if err := kubernetesutil.WaitForDeploymentToStabilize(context.Background(), client, ns.Name, d, 10*time.Minute); err != nil {
 					t.Fatalf("Timed out waiting for deployment to stabilize")
 				}
 				if testCase.deploymentValidation != nil {
@@ -290,7 +291,7 @@ func TestDev(t *testing.T) {
 			}()
 
 			for _, j := range testCase.jobs {
-				if err := kubernetesutil.WaitForJobToStabilize(client, ns.Name, j, 10*time.Minute); err != nil {
+				if err := kubernetesutil.WaitForJobToStabilize(context.Background(), client, ns.Name, j, 10*time.Minute); err != nil {
 					t.Fatalf("Timed out waiting for job to stabilize")
 				}
 				if testCase.jobValidation != nil {

pkg/skaffold/build/kaniko/kaniko.go

@@ -29,7 +29,7 @@ import (
 
 // Build builds a list of artifacts with Kaniko.
 func (b *Builder) Build(ctx context.Context, out io.Writer, tagger tag.Tagger, artifacts []*latest.Artifact) ([]build.Artifact, error) {
-	teardown, err := b.setupSecret()
+	teardown, err := b.setupSecret(out)
 	if err != nil {
 		return nil, errors.Wrap(err, "setting up secret")
 	}
@@ -39,7 +39,7 @@ func (b *Builder) Build(ctx context.Context, out io.Writer, tagger tag.Tagger, a
 }
 
 func (b *Builder) buildArtifact(ctx context.Context, out io.Writer, tagger tag.Tagger, artifact *latest.Artifact) (string, error) {
-	initialTag, err := runKaniko(ctx, out, artifact, b.KanikoBuild)
+	initialTag, err := b.run(ctx, out, artifact, b.KanikoBuild)
 	if err != nil {
 		return "", errors.Wrapf(err, "kaniko build for [%s]", artifact.ImageName)
 	}

pkg/skaffold/build/kaniko/logs.go

@@ -0,0 +1,68 @@
+/*
+Copyright 2018 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 kaniko
+
+import (
+	"io"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/constants"
+	"github.com/sirupsen/logrus"
+	"k8s.io/api/core/v1"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+)
+
+// logLevel makes sure kaniko logs at least at Info level.
+func logLevel() logrus.Level {
+	level := logrus.GetLevel()
+	if level < logrus.InfoLevel {
+		return logrus.InfoLevel
+	}
+	return level
+}
+
+func streamLogs(out io.Writer, name string, pods corev1.PodInterface) func() {
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	var retry int32 = 1
+	go func() {
+		defer wg.Done()
+
+		for atomic.LoadInt32(&retry) == 1 {
+			r, err := pods.GetLogs(name, &v1.PodLogOptions{
+				Follow:    true,
+				Container: constants.DefaultKanikoContainerName,
+			}).Stream()
+			if err != nil {
+				logrus.Debugln("unable to get kaniko pod logs:", err)
+				time.Sleep(1 * time.Second)
+				continue
+			}
+
+			io.Copy(out, r)
+			return
+		}
+	}()
+
+	return func() {
+		atomic.StoreInt32(&retry, 0)
+		wg.Wait()
+	}
+}

pkg/skaffold/build/kaniko/logs_test.go

@@ -0,0 +1,49 @@
+/*
+Copyright 2018 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 kaniko
+
+import (
+	"testing"
+
+	"github.com/GoogleContainerTools/skaffold/testutil"
+
+	"github.com/sirupsen/logrus"
+)
+
+func TestLogLevel(t *testing.T) {
+	defer func(l logrus.Level) { logrus.SetLevel(l) }(logrus.GetLevel())
+
+	tests := []struct {
+		logrusLevel logrus.Level
+		expected    logrus.Level
+	}{
+		{logrusLevel: logrus.DebugLevel, expected: logrus.DebugLevel},
+		{logrusLevel: logrus.InfoLevel, expected: logrus.InfoLevel},
+		{logrusLevel: logrus.WarnLevel, expected: logrus.InfoLevel},
+		{logrusLevel: logrus.ErrorLevel, expected: logrus.InfoLevel},
+		{logrusLevel: logrus.FatalLevel, expected: logrus.InfoLevel},
+		{logrusLevel: logrus.PanicLevel, expected: logrus.InfoLevel},
+	}
+
+	for _, test := range tests {
+		logrus.SetLevel(test.logrusLevel)
+
+		kanikoLevel := logLevel()
+
+		testutil.CheckDeepEqual(t, test.expected, kanikoLevel)
+	}
+}

pkg/skaffold/build/kaniko/run.go

@@ -20,61 +20,46 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"sync"
-	"sync/atomic"
-	"time"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/build/kaniko/sources"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/constants"
-
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/docker"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
-	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
-func runKaniko(ctx context.Context, out io.Writer, artifact *latest.Artifact, cfg *latest.KanikoBuild) (string, error) {
+func (b *Builder) run(ctx context.Context, out io.Writer, artifact *latest.Artifact, cfg *latest.KanikoBuild) (string, error) {
 	initialTag := util.RandomID()
-	s, err := sources.Retrieve(cfg)
-	if err != nil {
-		return "", errors.Wrap(err, "retrieving build context")
-	}
-	context, err := s.Setup(ctx, artifact, cfg, initialTag)
+
+	s := sources.Retrieve(cfg)
+	context, err := s.Setup(ctx, out, artifact, initialTag)
 	if err != nil {
 		return "", errors.Wrap(err, "setting up build context")
 	}
-	defer s.Cleanup(ctx, cfg)
-	dockerfilePath := artifact.DockerArtifact.DockerfilePath
+	defer s.Cleanup(ctx)
 
 	client, err := kubernetes.GetClientset()
 	if err != nil {
 		return "", errors.Wrap(err, "")
 	}
-	pods := client.CoreV1().Pods(cfg.Namespace)
 
 	imageDst := fmt.Sprintf("%s:%s", artifact.ImageName, initialTag)
 	args := []string{
-		fmt.Sprintf("--dockerfile=%s", dockerfilePath),
+		fmt.Sprintf("--dockerfile=%s", artifact.DockerArtifact.DockerfilePath),
 		fmt.Sprintf("--context=%s", context),
 		fmt.Sprintf("--destination=%s", imageDst),
-		fmt.Sprintf("-v=%s", logrus.GetLevel().String()),
+		fmt.Sprintf("-v=%s", logLevel().String()),
 	}
 	args = append(args, docker.GetBuildArgs(artifact.DockerArtifact)...)
 
-	p, err := pods.Create(s.Pod(cfg, args))
+	pods := client.CoreV1().Pods(cfg.Namespace)
+	p, err := pods.Create(s.Pod(args))
 	if err != nil {
 		return "", errors.Wrap(err, "creating kaniko pod")
 	}
-	if err := s.ModifyPod(p); err != nil {
-		return "", errors.Wrap(err, "modifying kaniko pod")
-	}
-	waitForLogs := streamLogs(out, p.Name, pods)
-
 	defer func() {
 		if err := pods.Delete(p.Name, &metav1.DeleteOptions{
 			GracePeriodSeconds: new(int64),
@@ -83,45 +68,17 @@ func runKaniko(ctx context.Context, out io.Writer, artifact *latest.Artifact, cf
 		}
 	}()
 
-	timeout, err := time.ParseDuration(cfg.Timeout)
-	if err != nil {
-		return "", errors.Wrap(err, "parsing timeout")
+	if err := s.ModifyPod(ctx, p); err != nil {
+		return "", errors.Wrap(err, "modifying kaniko pod")
 	}
 
-	if err := kubernetes.WaitForPodComplete(pods, p.Name, timeout); err != nil {
+	waitForLogs := streamLogs(out, p.Name, pods)
+
+	if err := kubernetes.WaitForPodComplete(ctx, pods, p.Name, b.timeout); err != nil {
 		return "", errors.Wrap(err, "waiting for pod to complete")
 	}
 
 	waitForLogs()
 
 	return imageDst, nil
 }
-
-func streamLogs(out io.Writer, name string, pods corev1.PodInterface) func() {
-	var wg sync.WaitGroup
-	wg.Add(1)
-
-	var retry int32 = 1
-	go func() {
-		defer wg.Done()
-
-		for atomic.LoadInt32(&retry) == 1 {
-			r, err := pods.GetLogs(name, &v1.PodLogOptions{
-				Follow:    true,
-				Container: constants.DefaultKanikoContainerName,
-			}).Stream()
-			if err == nil {
-				io.Copy(out, r)
-				return
-			}
-
-			logrus.Debugln("unable to get kaniko pod logs:", err)
-			time.Sleep(1 * time.Second)
-		}
-	}()
-
-	return func() {
-		atomic.StoreInt32(&retry, 0)
-		wg.Wait()
-	}
-}

pkg/skaffold/build/kaniko/secret.go

@@ -17,8 +17,10 @@ limitations under the License.
 package kaniko
 
 import (
+	"io"
 	"io/ioutil"
 
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/color"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/constants"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
 	"github.com/pkg/errors"
@@ -27,8 +29,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func (b *Builder) setupSecret() (func(), error) {
-	logrus.Debug("Creating kaniko secret")
+func (b *Builder) setupSecret(out io.Writer) (func(), error) {
+	color.Default.Fprintf(out, "Creating kaniko secret [%s]...\n", b.PullSecretName)
 
 	client, err := kubernetes.GetClientset()
 	if err != nil {

pkg/skaffold/build/kaniko/sources/gcs.go

@@ -19,23 +19,25 @@ package sources
 import (
 	"context"
 	"fmt"
+	"io"
 
 	cstorage "cloud.google.com/go/storage"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/color"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/docker"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/gcp"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest"
 	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
 )
 
 type GCSBucket struct {
+	cfg     *latest.KanikoBuild
 	tarName string
 }
 
 // Setup uploads the context to the provided GCS bucket
-func (g *GCSBucket) Setup(ctx context.Context, artifact *latest.Artifact, cfg *latest.KanikoBuild, initialTag string) (string, error) {
-	bucket := cfg.BuildContext.GCSBucket
+func (g *GCSBucket) Setup(ctx context.Context, out io.Writer, artifact *latest.Artifact, initialTag string) (string, error) {
+	bucket := g.cfg.BuildContext.GCSBucket
 	if bucket == "" {
 		guessedProjectID, err := gcp.ExtractProjectID(artifact.ImageName)
 		if err != nil {
@@ -44,32 +46,35 @@ func (g *GCSBucket) Setup(ctx context.Context, artifact *latest.Artifact, cfg *l
 
 		bucket = guessedProjectID
 	}
-	logrus.Debugln("Upload sources to", bucket, "GCS bucket")
+
+	color.Default.Fprintln(out, "Uploading sources to", bucket, "GCS bucket")
 
 	g.tarName = fmt.Sprintf("context-%s.tar.gz", initialTag)
 	if err := docker.UploadContextToGCS(ctx, artifact.Workspace, artifact.DockerArtifact, bucket, g.tarName); err != nil {
 		return "", errors.Wrap(err, "uploading sources to GCS")
 	}
-	context := fmt.Sprintf("gs://%s/%s", cfg.BuildContext.GCSBucket, g.tarName)
+
+	context := fmt.Sprintf("gs://%s/%s", g.cfg.BuildContext.GCSBucket, g.tarName)
 	return context, nil
 }
 
 // Pod returns the pod template for this builder
-func (g *GCSBucket) Pod(cfg *latest.KanikoBuild, args []string) *v1.Pod {
-	return podTemplate(cfg, args)
+func (g *GCSBucket) Pod(args []string) *v1.Pod {
+	return podTemplate(g.cfg, args)
 }
 
 // ModifyPod does nothing here, since we just need to let kaniko run to completion
-func (g *GCSBucket) ModifyPod(p *v1.Pod) error {
+func (g *GCSBucket) ModifyPod(ctx context.Context, p *v1.Pod) error {
 	return nil
 }
 
 // Cleanup deletes the tarball from the GCS bucket
-func (g *GCSBucket) Cleanup(ctx context.Context, cfg *latest.KanikoBuild) error {
+func (g *GCSBucket) Cleanup(ctx context.Context) error {
 	c, err := cstorage.NewClient(ctx)
 	if err != nil {
 		return err
 	}
 	defer c.Close()
-	return c.Bucket(cfg.BuildContext.GCSBucket).Object(g.tarName).Delete(ctx)
+
+	return c.Bucket(g.cfg.BuildContext.GCSBucket).Object(g.tarName).Delete(ctx)
 }