<file_sep>package main

import (
	"database/sql"
	"fmt"
	_ "github.com/go-sql-driver/mysql"
	appsv1 "k8s.io/api/apps/v1"
	"strings"
)

const (
	username = "root"
	password = "<PASSWORD>"
	ip       = "127.0.0.1"
	port     = "30944"
	dbName   = "RUNOOB"
)

var DB *sql.DB

func InitDB() error {
	path := strings.Join([]string{username, ":", password, "@tcp(", ip, ":", port, ")/", dbName, "?charset=utf8"}, "")
	DB, _ = sql.Open("mysql", path)
	DB.SetConnMaxLifetime(100)
	DB.SetMaxIdleConns(10)

	if err := DB.Ping(); err != nil {
		fmt.Printf("opon database fail: %v\n", err)
		return err
	}
	fmt.Println("connnect success")
	return nil
}

func InserDeployment(dp *appsv1.Deployment) bool {
	tx, err := DB.Begin()
	if err != nil {
		fmt.Println("tx fail")
		return false
	}
	stmt, err := tx.Prepare("INSERT INTO deployment_test (`deployment_name`, `deployment_namespace`, `desired_replicas`, `ready_replicas`) VALUES (?, ?, ?, ?)")
	if err != nil {
		fmt.Println("Prepare fail")
		return false
	}
	res, err := stmt.Exec(dp.Name, dp.Namespace, dp.Spec.Replicas, dp.Status.ReadyReplicas)
	if err != nil {
		fmt.Println("Exec fail")
		return false
	}
	tx.Commit()
	fmt.Println(res.LastInsertId())
	return true

}

func UpdateDeployment(dp *appsv1.Deployment) bool {
	tx, err := DB.Begin()
	if err != nil {
		fmt.Println("tx fail")
	}

	stmt, err := tx.Prepare("UPDATE deployment_test SET desired_replicas = ? , ready_replicas = ? WHERE deployment_name = ?")

	if err != nil {
		fmt.Println("Prepare fail")
		return false
	}

	res, err := stmt.Exec(dp.Spec.Replicas, dp.Status.ReadyReplicas, dp.Name)
	if err != nil {
		fmt.Println("Exec fail")
		return false
	}

	tx.Commit()
	fmt.Println(res.LastInsertId())
	return true
}

func DeleteDeployment(key string) bool {
	strs := strings.Split(key, "/")
	name := strs[1]
	tx, err := DB.Begin()
	if err != nil {
		fmt.Println("tx fail")
	}
	fmt.Println("The key is:", name)
	stmt, err := tx.Prepare("DELETE FROM deployment_test WHERE deployment_name = ?")

	if err != nil {
		fmt.Println("Prepare fail")
		return false
	}

	res, err := stmt.Exec(name)
	if err != nil {
		fmt.Println("Exec fail")
		return false
	}

	tx.Commit()
	fmt.Println(res.LastInsertId())
	return true
}
<file_sep>package main

import (
	"fmt"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	appsinformers "k8s.io/client-go/informers/apps/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
	appslisters "k8s.io/client-go/listers/apps/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/klog"
)

const controllerAgentName = "deployment-controller"

type Delta struct {
	key   string
	event string
}

type Controller struct {
	kubeclientset     kubernetes.Interface
	deploymentsLister appslisters.DeploymentLister
	deploymentsSynced cache.InformerSynced

	workqueue workqueue.RateLimitingInterface
	recorder  record.EventRecorder
}

func NewController(
	kubeclientset kubernetes.Interface,
	deploymentInformer appsinformers.DeploymentInformer) *Controller {

	klog.V(4).Info("Creating event broadcaster")
	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartLogging(klog.Infof)
	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events("")})
	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})

	controller := &Controller{
		kubeclientset:     kubeclientset,
		deploymentsLister: deploymentInformer.Lister(),
		deploymentsSynced: deploymentInformer.Informer().HasSynced,
		workqueue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Foos"),
		recorder:          recorder,
	}

	klog.Info("Setting up event handlers")

	deploymentInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleAdd,
		UpdateFunc: func(old, new interface{}) {
			newDepl := new.(*appsv1.Deployment)
			oldDepl := old.(*appsv1.Deployment)
			if newDepl.ResourceVersion == oldDepl.ResourceVersion {
				return
			}
			controller.handleUpdate(new)
		},
		DeleteFunc: controller.handleDelete,
	})
	return controller
}

func (c *Controller) handleAdd(obj interface{}) {
	var key string
	var err error
	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
		utilruntime.HandleError(err)
		return
	}
	d := Delta{key: key, event: "Add"}
	c.workqueue.Add(d)
}

func (c *Controller) handleUpdate(obj interface{}) {
	var key string
	var err error
	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
		utilruntime.HandleError(err)
		return
	}
	d := Delta{key: key, event: "Update"}
	c.workqueue.Add(d)
}

func (c *Controller) handleDelete(obj interface{}) {
	var key string
	var err error
	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
		utilruntime.HandleError(err)
		return
	}
	d := Delta{key: key, event: "Delete"}
	c.workqueue.Add(d)
}

func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
	defer utilruntime.HandleCrash()
	defer c.workqueue.ShutDown()

	// Start the informer factories to begin populating the informer caches
	klog.Info("Starting Foo controller")

	// Wait for the caches to be synced before starting workers
	klog.Info("Waiting for informer caches to sync")
	if ok := cache.WaitForCacheSync(stopCh, c.deploymentsSynced); !ok {
		return fmt.Errorf("failed to wait for caches to sync")
	}

	klog.Info("Starting workers")
	// Launch two workers to process Foo resources
	for i := 0; i < threadiness; i++ {
		go wait.Until(c.runWorker, time.Second, stopCh)
	}

	klog.Info("Started workers")
	<-stopCh
	klog.Info("Shutting down workers")

	return nil
}

func (c *Controller) runWorker() {
	for c.processNextWorkItem() {

	}
}

func (c *Controller) processNextWorkItem() bool {
	obj, shutdown := c.workqueue.Get()

	if shutdown {
		return false
	}

	err := func(obj interface{}) error {
		defer c.workqueue.Done(obj)

		// handle logic
		var d Delta
		var ok bool

		if d, ok = obj.(Delta); !ok {
			c.workqueue.Forget(obj)
			utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
			return nil
		}

		namespace, name, err := cache.SplitMetaNamespaceKey(d.key)
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", d.key))
			return nil
		}
		dp, err := c.deploymentsLister.Deployments(namespace).Get(name)
		if err != nil && d.event != "Delete" {
			if errors.IsNotFound(err) {
				utilruntime.HandleError(fmt.Errorf("deployment '%s' in work queue no longer exists", d.key))
				return nil
			}
			return err
		}
		if dp != nil {
			fmt.Println("This is the output", dp.Name, dp.Status.ReadyReplicas, dp.Status.Replicas, dp.Namespace)
			if d.event == "Add" {
				InserDeployment(dp)
			} else if d.event == "Update" {
				UpdateDeployment(dp)
			}
		} else {
			DeleteDeployment(d.key)
		}

		c.workqueue.Forget(obj)
		klog.Infof("Successfully synced")
		return nil
	}(obj)

	if err != nil {
		utilruntime.HandleError(err)
		return true
	}

	return true
}
<file_sep># deployment-controller
Deployment controller 目前的功能是同步数据到MySQL
