use k8s exec to get workload config from workers, this will feed up swarm state

go.mod

@@ -5,9 +5,7 @@ go 1.17
 require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/fsnotify/fsnotify v1.5.1
-	github.com/google/go-cmp v0.5.7
 	github.com/gorilla/mux v1.8.0
-	github.com/mitchellh/go-homedir v1.1.0
 	github.com/mitchellh/mapstructure v1.4.3
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/cobra v1.4.0
@@ -33,6 +31,7 @@ require (
 	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/go-cmp v0.5.7 // indirect
 	github.com/google/gofuzz v1.1.0 // indirect
 	github.com/googleapis/gnostic v0.5.5 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
@@ -42,6 +41,8 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/magiconair/properties v1.8.5 // indirect
 	github.com/mailru/easyjson v0.7.6 // indirect
+	github.com/mitchellh/go-homedir v1.1.0 // 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/pelletier/go-toml v1.9.4 // indirect

go.sum

@@ -398,6 +398,7 @@ github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh
 github.com/mitchellh/mapstructure v1.4.1/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/mitchellh/mapstructure v1.4.3 h1:OVowDSCllw/YjdLkam3/sm7wEtOy59d8ndGgCcyj8cs=
 github.com/mitchellh/mapstructure v1.4.3/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
+github.com/moby/spdystream v0.2.0 h1:cjW1zVyyoiM0T7b6UoySUFqzXMoqRckQtXwGPiBhOM8=
 github.com/moby/spdystream v0.2.0/go.mod h1:f7i0iNDQJ059oMTcWxx8MA/zKFIuD/lY+0GqbN2Wy8c=
 github.com/moby/term v0.0.0-20210610120745-9d4ed1856297/go.mod h1:vgPCkQMyxTZ7IDy8SXRufE172gr8+K/JE/7hHFxHW3A=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=

pkg/config/config.go

@@ -3,8 +3,11 @@ package config
 import (
 	"fmt"
 	logger "github.com/marcosQuesada/k8s-lab/pkg/log"
+	"github.com/mitchellh/mapstructure"
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
+	"gopkg.in/yaml.v3"
 	"os"
 )
 
@@ -118,6 +121,20 @@ func (a *Workload) Difference(newWorkload *Workload) (included, excluded []Job)
 	return
 }
 
+func Decode(raw []byte) (*Workloads, error) {
+	data := make(map[interface{}]interface{})
+	if err := yaml.Unmarshal(raw, &data); err != nil {
+		return nil, errors.Wrap(err, "unable to unmarshall")
+	}
+
+	c := &Workloads{}
+	if err := mapstructure.Decode(data, c); err != nil {
+		return nil, errors.Wrap(err, "unable to decode")
+	}
+
+	return c, nil
+}
+
 func toMap(set []Job) map[string]struct{} {
 	res := map[string]struct{}{}
 	for _, s := range set {

pkg/operator/configmap/provider_test.go

@@ -23,7 +23,7 @@ func TestNewProvider_ItUpdatesConfigMapOnAssignWorkload(t *testing.T) {
 	}
 	spew.Dump(cm.Data)
 
-	p := NewProvider(clientset, namespace, configMapName)
+	p := NewProvider(clientset)
 	w := &config.Workloads{
 		Version: 1,
 		Workloads: map[string]*config.Workload{
@@ -33,7 +33,7 @@ func TestNewProvider_ItUpdatesConfigMapOnAssignWorkload(t *testing.T) {
 		},
 	}
 
-	if err := p.Set(context.Background(), w); err != nil {
+	if err := p.Set(context.Background(), namespace, configMapName, w); err != nil {
 		t.Fatalf("unexepcted error setting workload %v, got %v", w, err)
 	}
 }
@@ -52,8 +52,8 @@ func TestNewProvider_ItGetsWorkloadsFromConfigMap(t *testing.T) {
 	}
 	spew.Dump(cm.Data)
 
-	p := NewProvider(clientset, namespace, configMapName)
-	w, err := p.Get(context.Background())
+	p := NewProvider(clientset)
+	w, err := p.Get(context.Background(), namespace, configMapName)
 	if err != nil {
 		t.Fatalf("unexepcted error setting workload %v, got %v", w, err)
 	}

pkg/operator/controller_test.go

@@ -23,7 +23,8 @@ func TestController_ItGetsCreatedOnListeningPodsWithPodAddition(t *testing.T) {
 	cl := fake.NewSimpleClientset(p)
 	i := informers.NewSharedInformerFactory(cl, 0)
 	pi := i.Core().V1().Pods()
-	ctl := New(eh, pi.Informer(), "Pod")
+	fr := &fakeRunner{}
+	ctl := New(eh, pi.Informer(), fr, "Pod")
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -89,7 +90,8 @@ func TestController_ItGetsDeletedOnListeningPodsWithPodAdditionWithoutBeingPrelo
 	cl := fake.NewSimpleClientset()
 	i := informers.NewSharedInformerFactory(cl, 0)
 	pi := i.Core().V1().Pods()
-	ctl := New(eh, pi.Informer(), "Pod")
+	fr := &fakeRunner{}
+	ctl := New(eh, pi.Informer(), fr, "Pod")
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -121,15 +123,21 @@ type fakeHandler struct {
 	totalDeleted int32
 }
 
-func (f *fakeHandler) Handle(ctx context.Context, o runtime.Object) error {
+func (f *fakeHandler) Create(ctx context.Context, o runtime.Object) error {
 	atomic.AddInt32(&f.totalCreated, 1)
 	return nil
 }
 
-func (f *fakeHandler) Delete(ctx context.Context, namespace, name string) error {
+func (f *fakeHandler) Update(ctx context.Context, o, n runtime.Object) error {
+	atomic.AddInt32(&f.totalCreated, 1)
+	return nil
+}
+
+func (f *fakeHandler) Delete(ctx context.Context, o runtime.Object) error {
 	atomic.AddInt32(&f.totalDeleted, 1)
 	return nil
 }
+
 func (f *fakeHandler) created() int32 {
 	return atomic.LoadInt32(&f.totalCreated)
 }
@@ -160,3 +168,13 @@ func getFakePod(namespace, name string) *apiv1.Pod {
 		},
 	}
 }
+
+type fakeRunner struct{}
+
+func (f fakeRunner) Process(e interface{}) {
+
+}
+
+func (f fakeRunner) Run(ctx context.Context, h func(context.Context, interface{}) error) {
+
+}

pkg/operator/pod/exec.go

@@ -0,0 +1,54 @@
+package pod
+
+import (
+	"bytes"
+	"fmt"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
+
+	"k8s.io/client-go/kubernetes/scheme"
+	restclient "k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/remotecommand"
+)
+
+// ExecCmd exec command on specific pod and wait the command's output.
+func ExecCmd(client kubernetes.Interface, config *restclient.Config, namespace, name string,
+	command string) ([]byte, error) {
+	cmd := []string{
+		"sh",
+		"-c",
+		command,
+	}
+	req := client.CoreV1().RESTClient().Post().Resource("pods").Name(name).
+		Namespace(namespace).SubResource("exec")
+	option := &v1.PodExecOptions{
+		Command: cmd,
+		Stdin:   false,
+		Stdout:  true,
+		Stderr:  true,
+		TTY:     true,
+	}
+	req.VersionedParams(
+		option,
+		scheme.ParameterCodec,
+	)
+	exec, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
+	if err != nil {
+		return nil, err
+	}
+
+	outBuf := &bytes.Buffer{}
+	errBuf := &bytes.Buffer{}
+	err = exec.Stream(remotecommand.StreamOptions{
+		Stdout: outBuf,
+		Stderr: errBuf,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	if errBuf.Len() != 0 {
+		return nil, fmt.Errorf("StdErr error %s", errBuf.String())
+	}
+	return outBuf.Bytes(), nil
+}

pkg/operator/pod/exec_test.go

@@ -0,0 +1,34 @@
+package pod
+
+import (
+	"github.com/marcosQuesada/k8s-lab/pkg/operator"
+	"k8s.io/client-go/tools/clientcmd"
+	"os"
+	"testing"
+)
+
+// @TODO: Think on how to test it
+func TestExecCmdExample(t *testing.T) {
+	t.Skip()
+	clientSet := operator.BuildExternalClient()
+	kubeConfigPath := os.Getenv("HOME") + "/.kube/config"
+	config, err := clientcmd.BuildConfigFromFlags("", kubeConfigPath)
+	if err != nil {
+		t.Fatalf("unable to get cluster config from flags, error %v", err)
+	}
+
+	namespace := "swarm"
+	name := "swarm-worker-0"
+	cmd := "cat /app/config/config.yml"
+	r, err := ExecCmd(clientSet, config, namespace, name, cmd)
+	if err != nil {
+		t.Fatalf("unable to execute command, error %v", err)
+	}
+
+	//spew.Dump(outBuf.String(), errBuf.String())
+
+	if err := os.WriteFile("foo.yaml", r, 0644); err != nil {
+		t.Fatalf("unable to write fil, error %v", err)
+	}
+
+}

pkg/operator/resource_event_handler_test.go

@@ -13,9 +13,9 @@ func TestResourceEventHandler_AddPodIncludesItemOnQueue(t *testing.T) {
 	name := "swarm-worker-0"
 	namespace := "swarm"
 	q := newFakeQueue()
-	reh := NewResourceEventHandler(q)
+	reh := NewResourceEventHandler()
 	p := getFakePod(namespace, name)
-	reh.Add(&p)
+	reh.Create(&p)
 
 	if expected, got := 1, q.Len(); expected != got {
 		t.Fatalf("unexpected queue size, expected %d got %d", expected, got)
@@ -38,9 +38,12 @@ func TestResourceEventHandler_UpdatePodIncludesItemOnQueue(t *testing.T) {
 	name := "swarm-worker-0"
 	namespace := "swarm"
 	q := newFakeQueue()
-	reh := NewResourceEventHandler(q)
+	reh := NewResourceEventHandler()
 	p := getFakePod(namespace, name)
-	reh.Update(&p, &p)
+	_, err := reh.Update(&p, &p) // @TODO: check event
+	if err != nil {
+		t.Fatalf("unexpected error updating resource %v", err)
+	}
 
 	if expected, got := 1, q.Len(); expected != got {
 		t.Fatalf("unexpected queue size, expected %d got %d", expected, got)
@@ -63,9 +66,12 @@ func TestResourceEventHandler_DeletePodIncludesItemOnQueue(t *testing.T) {
 	name := "swarm-worker-0"
 	namespace := "swarm"
 	q := newFakeQueue()
-	reh := NewResourceEventHandler(q)
+	reh := NewResourceEventHandler()
 	p := getFakePod(namespace, name)
-	reh.Delete(&p)
+	_, err := reh.Delete(&p) // @TODO: validate event
+	if err != nil {
+		t.Fatalf("unexpected error deleting resource %v", err)
+	}
 
 	if expected, got := 1, q.Len(); expected != got {
 		t.Fatalf("unexpected queue size, expected %d got %d", expected, got)

pkg/operator/runner_test.go

@@ -14,7 +14,7 @@ func TestItConsumesProducedEntriesWithSuccess(t *testing.T) {
 		atomic.AddInt32(&totalCalls, 1)
 		return nil
 	}
-	r := NewRunner()
+	r := NewRunner().(*runner)
 	r.workerFrequency = time.Millisecond * 50
 	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*100)
 	defer cancel()
@@ -34,7 +34,7 @@ func TestItRetriesConsumedEntriesOnHandlingErrorUntilMaxRetries(t *testing.T) {
 		atomic.AddInt32(&totalCalls, 1)
 		return errors.New("foo error")
 	}
-	r := NewRunner()
+	r := NewRunner().(*runner)
 	r.workerFrequency = time.Millisecond * 50
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

services/fake-worker/k8s/statefulset.yaml

@@ -9,7 +9,7 @@ spec:
   selector:
     matchLabels:
       app: swarm-worker
-  serviceName: swarm-worker
+  serviceName: swarm-worker-headless
   replicas: 2
   template:
     metadata:
@@ -61,6 +61,7 @@ apiVersion: v1
 kind: Service
 metadata:
   name: swarm-worker-headless
+  namespace: swarm
   labels:
     app: swarm-worker
 spec:
@@ -69,4 +70,4 @@ spec:
       name: http
   clusterIP: None
   selector:
-    app: swarm-worker
+    app: swarm-worker

services/swarm-pool-controller/cmd/external.go

@@ -13,11 +13,13 @@ import (
 	"github.com/marcosQuesada/k8s-lab/services/swarm-pool-controller/internal/infra/k8s/crd"
 	"github.com/marcosQuesada/k8s-lab/services/swarm-pool-controller/internal/infra/k8s/crd/apis/swarm/v1alpha1"
 	crdinformers "github.com/marcosQuesada/k8s-lab/services/swarm-pool-controller/internal/infra/k8s/crd/generated/informers/externalversions"
+	wp "github.com/marcosQuesada/k8s-lab/services/swarm-pool-controller/internal/infra/k8s/pod"
 	statefulset "github.com/marcosQuesada/k8s-lab/services/swarm-pool-controller/internal/infra/k8s/statefulset"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/clientcmd"
 	"net/http"
 	"os"
 	"os/signal"
@@ -66,7 +68,15 @@ var externalCmd = &cobra.Command{
 		appm := app.NewManager(ex, swl)
 		selSt := statefulset.NewSelectorStore()
 		pr := app.NewProvider(swl, stsl, podl)
-		ctl := app.NewSwarmController(swarmClientSet, selSt, appm, pr, operator.NewRunner())
+
+		kubeConfigPath := os.Getenv("HOME") + "/.kube/config"
+		restConfig, err := clientcmd.BuildConfigFromFlags("", kubeConfigPath)
+		if err != nil {
+			log.Fatalf("unable to get cluster config from flags, error %v", err)
+		}
+
+		wklpr := wp.NewProvider(clientSet, restConfig)
+		ctl := app.NewSwarmController(swarmClientSet, selSt, appm, pr, wklpr, operator.NewRunner())
 		go ctl.Run(ctx)
 
 		crdh := crd.NewHandler(ctl)