added support for topology creation

.github/workflows/cicd.yaml

@@ -15,9 +15,19 @@ jobs:
       - name: Build binary
         run: |
           go build -v .
-      - name: Run command
+      - name: Setup docker
+        uses: docker/setup-buildx-action@v2
+      - name: Setup kind cluster
+        run: |
+          go install sigs.k8s.io/kind@v0.20.0
+          kind create cluster --wait 5m
+      - name: Create topology
         run: |
           ./ktm -topo testdata/client-server.yaml
+      - name: Check topology
+        run: |
+          docker exec kind-control-plane kubectl wait --for=condition=Ready pods -n client-server --all
+          docker exec kind-control-plane kubectl get pods -A
       - name: Cleanup
         run: |
           ./ktm -cleanup

main.go

@@ -16,7 +16,9 @@ func main() {
 	if *cleanupPtr {
 		printUsage = false
 
-		log.Println("Cleaning up any active topology ...")
+		if err := topology.Cleanup(); err != nil {
+			log.Fatalf("Failed cleaning up: %v\n", err)
+		}
 	}
 
 	if *topoPtr != "" {

pkg/topology/api.go → pkg/api/api.go

@@ -1,4 +1,4 @@
-package topology
+package api
 
 type Topology struct {
 	Name  string `yaml:"name"`

pkg/k8s/client.go

@@ -0,0 +1,34 @@
+package k8s
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+type client struct {
+	config    *rest.Config
+	clientset *kubernetes.Clientset
+	ctx       context.Context
+}
+
+func NewClient() (*client, error) {
+	log.Println("Creating new K8S client ...")
+
+	config, err := clientcmd.BuildConfigFromFlags("", clientcmd.RecommendedHomeFile)
+	if err != nil {
+		return nil, fmt.Errorf("could not build kube config from flag: %v", err)
+	}
+
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return nil, fmt.Errorf("could not create new clientset: %v", err)
+	}
+
+	log.Println("Successfully created K8s client !")
+	return &client{clientset: clientset, config: config, ctx: context.TODO()}, nil
+}

pkg/k8s/namespace.go

@@ -0,0 +1,52 @@
+package k8s
+
+import (
+	"fmt"
+	"log"
+
+	apiv1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func (c *client) CreateNamespace(name string) error {
+	if name == "" {
+		return fmt.Errorf("namespace not provided")
+	}
+
+	log.Printf("Creating namespace %s ...\n", name)
+	_, err := c.clientset.CoreV1().Namespaces().Create(
+		c.ctx, &apiv1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:   name,
+				Labels: map[string]string{"owner": "ktm"},
+			},
+		},
+		metav1.CreateOptions{},
+	)
+
+	if err != nil {
+		return fmt.Errorf("could not create namespace %s: %v", name, err)
+	}
+	log.Println("Successfully created namespace !")
+
+	return nil
+}
+
+func (c *client) DeleteNamespace(name string) error {
+	if name == "" {
+		return fmt.Errorf("namespace not provided")
+	}
+
+	log.Printf("Deleting namespace %s ...\n", name)
+	err := c.clientset.CoreV1().Namespaces().Delete(
+		c.ctx, name,
+		metav1.DeleteOptions{},
+	)
+
+	if err != nil {
+		return fmt.Errorf("could not delete namespace %s: %v", name, err)
+	}
+	log.Println("Successfully deleted namespace !")
+
+	return nil
+}

pkg/k8s/node.go

@@ -0,0 +1,47 @@
+package k8s
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/ashutshkumr/ktm/pkg/api"
+	apiv1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func (c *client) CreateNode(ns string, n *api.Node) error {
+	if n == nil {
+		return fmt.Errorf("node not provided")
+	}
+
+	log.Printf("Creating node %s ...\n", n.Name)
+	_, err := c.clientset.CoreV1().Pods(ns).Create(
+		c.ctx,
+		&apiv1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      n.Name,
+				Namespace: ns,
+				Labels:    map[string]string{"owner": "ktm"},
+			},
+			Spec: apiv1.PodSpec{
+				Containers: []apiv1.Container{
+					{
+						Name:            n.Name + "-ctr",
+						Image:           n.Image,
+						Command:         n.Cmd,
+						ImagePullPolicy: apiv1.PullIfNotPresent,
+					},
+				},
+				RestartPolicy: apiv1.RestartPolicyAlways,
+			},
+		},
+		metav1.CreateOptions{},
+	)
+
+	if err != nil {
+		return fmt.Errorf("could not create node %s: %v", n.Name, err)
+	}
+	log.Println("Successfully created node !")
+
+	return nil
+}

pkg/topology/io.go

@@ -5,6 +5,7 @@ import (
 	"log"
 	"os"
 
+	"github.com/ashutshkumr/ktm/pkg/api"
 	"gopkg.in/yaml.v3"
 )
 
@@ -15,11 +16,12 @@ func NewFromFile(fileName string) error {
 		return fmt.Errorf("could not read topology file: %v", err)
 	}
 
-	t := Topology{}
+	t := api.Topology{}
 	log.Printf("Parsing topology ...\n%s\n", b)
 	if err := yaml.Unmarshal(b, &t); err != nil {
 		return fmt.Errorf("could not unmarshal topology file: %v", err)
 	}
+	log.Println("Successfully parsed topology !")
 
 	return New(&t)
 }

pkg/topology/k8s.go

@@ -1,5 +1,55 @@
 package topology
 
-func New(t *Topology) error {
+import (
+	"fmt"
+	"log"
+
+	"github.com/ashutshkumr/ktm/pkg/api"
+	"github.com/ashutshkumr/ktm/pkg/k8s"
+)
+
+func New(t *api.Topology) error {
+	if t == nil {
+		return fmt.Errorf("no topology provided")
+	}
+
+	log.Printf("Creating topology %s ...\n", t.Name)
+	k8sClient, err := k8s.NewClient()
+	if err != nil {
+		return fmt.Errorf("could not create k8s client: %v", err)
+	}
+
+	if err := k8sClient.CreateNamespace(t.Name); err != nil {
+		return err
+	}
+
+	for _, n := range t.Nodes {
+		if err := k8sClient.CreateNode(t.Name, &n); err != nil {
+			return err
+		}
+	}
+
+	log.Println("Successfully created topology !")
+	return nil
+}
+
+func Cleanup() error {
+	log.Println("Cleaning up any active topologies ...")
+	// k8sClient, err := k8s.NewClient()
+	// if err != nil {
+	// 	return fmt.Errorf("could not create k8s client: %v", err)
+	// }
+
+	// if err := k8sClient.CreateNamespace(t.Name); err != nil {
+	// 	return err
+	// }
+
+	// for _, n := range t.Nodes {
+	// 	if err := k8sClient.CreateNode(t.Name, &n); err != nil {
+	// 		return err
+	// 	}
+	// }
+
+	log.Println("Successfully created topology !")
 	return nil
 }