feat(pipeline/operator): k8s pod controller

Author: morlay

go.mod

@@ -3,21 +3,28 @@ module github.com/querycap/pipeline
 go 1.14
 
 require (
-	github.com/davecgh/go-spew v1.1.1
+	github.com/Microsoft/go-winio v0.4.14 // indirect
 	github.com/docker/distribution v2.7.1+incompatible // indirect
 	github.com/docker/docker v1.13.1
 	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/docker/go-units v0.4.0 // indirect
-	github.com/go-courier/courier v1.3.1
-	github.com/go-courier/httptransport v1.17.3
 	github.com/go-courier/semver v1.0.0
-	github.com/go-courier/snowflakeid v1.2.1
 	github.com/gomodule/redigo v2.0.0+incompatible
+	github.com/imdario/mergo v0.3.9 // indirect
 	github.com/minio/minio-go/v6 v6.0.55
 	github.com/onsi/gomega v1.9.0
 	github.com/opencontainers/go-digest v1.0.0-rc1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/sirupsen/logrus v1.6.0
 	github.com/spf13/afero v1.2.2
+	golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550 // indirect
+	golang.org/x/net v0.0.0-20200226121028-0de0cce0169b // indirect
+	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d // indirect
+	golang.org/x/sys v0.0.0-20200509044756-6aff5f38e54f // indirect
+	golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1 // indirect
+	golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 // indirect
 	gopkg.in/yaml.v2 v2.2.8
+	k8s.io/api v0.17.0
+	k8s.io/apimachinery v0.17.0
+	k8s.io/client-go v0.17.0
 )

pipeline/operator/container/image_registry.go

@@ -16,6 +16,7 @@ func ParseImageRegistry(uri string) (*ImageRegistry, error) {
 
 	endpoint := &ImageRegistry{}
 
+	endpoint.Name = u.Scheme
 	endpoint.Host = u.Host
 	endpoint.Prefix = u.Path
 
@@ -28,6 +29,7 @@ func ParseImageRegistry(uri string) (*ImageRegistry, error) {
 }
 
 type ImageRegistry struct {
+	Name     string
 	Host     string
 	Username string
 	Password string
@@ -43,3 +45,15 @@ func (s ImageRegistry) RegistryAuth() string {
 	b, _ := json.Marshal(authConfig)
 	return base64.StdEncoding.EncodeToString(b)
 }
+
+func (s ImageRegistry) DockerConfigJSON() []byte {
+	v := struct {
+		Auths map[string]types.AuthConfig `json:"auths"`
+	}{
+		Auths: map[string]types.AuthConfig{
+			s.Host: {Username: s.Username, Password: s.Password},
+		},
+	}
+	b, _ := json.Marshal(v)
+	return b
+}

pipeline/operator/container/operatormgr.go renamed to pipeline/operator/container/operator_mgr.go

@@ -7,26 +7,26 @@ import (
 	"github.com/querycap/pipeline/spec"
 )
 
-func NewOperatorMgr(podController PodController, container *Container) pipeline.OperatorMgr {
+func NewOperatorMgr(podController PodController, envs map[string]string) pipeline.OperatorMgr {
 	return &operatorMgr{
 		podController: podController,
-		container:     container,
+		envs:          envs,
 	}
 }
 
 type operatorMgr struct {
 	podController PodController
-	container     *Container
+	envs          map[string]string
 }
 
-func (d *operatorMgr) Up(scope string, stage string, step spec.Stage, replicas int) error {
+func (d *operatorMgr) Up(scope string, stage string, step spec.Stage, replicas int32) error {
 	c := Container{
 		Container: step.Container,
 		Image:     step.Uses.RefID(),
 		Replicas:  replicas,
 	}
 
-	c.Envs = c.Envs.Merge(d.container.Envs)
+	c.Envs = c.Envs.Merge(d.envs)
 
 	c.Envs["PIPELINE_SCOPE"] = scope
 	c.Envs["PIPELINE_STAGE"] = stage

pipeline/operator/container/operatormgr_test.go renamed to pipeline/operator/container/operator_mgr_test.go

@@ -28,7 +28,7 @@ func RunAll(t *testing.T, c PodController) {
 	s.Uses.Name = "nginx"
 	s.Uses.Version = *semver.MustParseVersion("1.17.10")
 
-	mgr := NewOperatorMgr(c, &Container{})
+	mgr := NewOperatorMgr(c, map[string]string{})
 
 	t.Run("start", func(t *testing.T) {
 		if err := mgr.Up(scope, stage, s, 3); err != nil {

pipeline/operator/container/pod_controller.go

@@ -8,7 +8,8 @@ import (
 )
 
 func PodNameByScopeAndStage(scope string, stage string) string {
-	return strings.Replace(scope+"--"+stage, ":", "__", -1)
+	parts := strings.Split(scope, "/")
+	return "p" + parts[len(parts)-1] + "-" + stage
 }
 
 type PodController interface {
@@ -19,6 +20,6 @@ type PodController interface {
 type Container struct {
 	spec.Container
 	Image       string
-	Replicas    int
+	Replicas    int32
 	Annotations map[string]string
 }

pipeline/operator/container/pod_controller__docker.go

@@ -35,27 +35,27 @@ type DockerPodController struct {
 func (c *DockerPodController) Apply(ctx context.Context, name string, container *Container) error {
 	imageRef := c.imageRegistry.Fix(container.Image)
 
-	if err := c.PullImageIfNotExists(ctx, imageRef); err != nil {
+	if err := c.pullImageIfNotExists(ctx, imageRef); err != nil {
 		return err
 	}
 
 	container.Annotations = map[string]string{
 		"pipeline": name,
 	}
 
-	list, err := c.ListMatchedRunningContainer(ctx, name)
+	list, err := c.listMatchedRunningContainer(ctx, name)
 	if err != nil {
 		return err
 	}
 
 	current := len(list)
 
-	offset := current - container.Replicas
+	offset := current - int(container.Replicas)
 
 	// scale up
 	if offset < 0 {
 		for i := 0; i < -offset; i++ {
-			if err := c.RunContainer(ctx, imageRef, container); err != nil {
+			if err := c.runContainer(ctx, imageRef, container); err != nil {
 				return err
 			}
 		}
@@ -67,7 +67,7 @@ func (c *DockerPodController) Apply(ctx context.Context, name string, container
 		rand.Shuffle(len(list), func(i, j int) { list[i], list[j] = list[j], list[i] })
 
 		for _, item := range list[0:offset] {
-			if err := c.KillContainer(ctx, item.ID); err != nil {
+			if err := c.killContainer(ctx, item.ID); err != nil {
 				return err
 			}
 		}
@@ -77,33 +77,35 @@ func (c *DockerPodController) Apply(ctx context.Context, name string, container
 }
 
 func (c *DockerPodController) Kill(ctx context.Context, name string) error {
-	list, err := c.ListMatchedRunningContainer(ctx, name)
+	list, err := c.listMatchedRunningContainer(ctx, name)
 	if err != nil {
 		return err
 	}
 
 	for i := range list {
-		if err := c.KillContainer(ctx, list[i].ID); err != nil {
+		if err := c.killContainer(ctx, list[i].ID); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
 
-func (c *DockerPodController) ListMatchedRunningContainer(ctx context.Context, name string) ([]types.Container, error) {
+func (c *DockerPodController) listMatchedRunningContainer(ctx context.Context, name string) ([]types.Container, error) {
 	containerListFilters := filters.NewArgs()
 	containerListFilters.Add("label", "pipeline="+name)
 
-	return c.ListRunningContainer(ctx, containerListFilters)
+	return c.listRunningContainer(ctx, containerListFilters)
 }
 
-func (c *DockerPodController) RunContainer(ctx context.Context, image string, cc *Container) error {
+func (c *DockerPodController) runContainer(ctx context.Context, image string, cc *Container) error {
 	logrus.WithContext(ctx).Debugf("running from %s", image)
 
 	containerConfig := &container.Config{
-		Image:  image,
-		Labels: cc.Annotations,
+		Image:      image,
+		Labels:     cc.Annotations,
+		Entrypoint: cc.Command,
+		Cmd:        cc.Args,
 	}
 
 	for k := range cc.Envs {
@@ -143,20 +145,20 @@ func (c *DockerPodController) RunContainer(ctx context.Context, image string, cc
 	return nil
 }
 
-func (c *DockerPodController) ListRunningContainer(ctx context.Context, args filters.Args) ([]types.Container, error) {
+func (c *DockerPodController) listRunningContainer(ctx context.Context, args filters.Args) ([]types.Container, error) {
 	return c.client.ContainerList(ctx, types.ContainerListOptions{
 		Filters: args,
 		All:     false,
 	})
 }
 
-func (c *DockerPodController) KillContainer(ctx context.Context, containerID string) error {
+func (c *DockerPodController) killContainer(ctx context.Context, containerID string) error {
 	logrus.WithContext(ctx).Debugf("killing %s", containerID)
 
 	return c.client.ContainerKill(ctx, containerID, "SIGKILL")
 }
 
-func (c *DockerPodController) PullImageIfNotExists(ctx context.Context, image string) error {
+func (c *DockerPodController) pullImageIfNotExists(ctx context.Context, image string) error {
 	imageListFilters := filters.NewArgs()
 	imageListFilters.Add("reference", strings.Replace(image, "docker.io/library/", "", 1))