Merge pull request #18 from micro-in-cn/agent

Agent

.github/workflows/release.yml

@@ -96,3 +96,12 @@ jobs:
 
           docker push ${{ secrets.REGISTRY_USERNAME }}/micro:latest
           docker push ${{ secrets.REGISTRY_USERNAME }}/micro:${{ env.RELEASE_VERSION }}
+      - name: agent
+        run: |
+          make agent
+          docker login docker.io -u ${{ secrets.REGISTRY_USERNAME }} -p ${{ secrets.REGISTRY_PASSWORD }}
+          docker tag agent:latest ${{ secrets.REGISTRY_USERNAME }}/agent:latest
+          docker tag agent:latest ${{ secrets.REGISTRY_USERNAME }}/agent:${{ env.RELEASE_VERSION }}
+
+          docker push ${{ secrets.REGISTRY_USERNAME }}/agent:latest
+          docker push ${{ secrets.REGISTRY_USERNAME }}/agent:${{ env.RELEASE_VERSION }}

.gitignore

@@ -14,3 +14,4 @@
 .idea/
 
 deployments/docker-compose/.env
+agent/config.json*

Makefile

@@ -17,3 +17,7 @@ config-srv:
 .PHONY: micro
 micro:
 	docker build -f deployments/docker/micro/Dockerfile . -t micro
+
+.PHONY: agent
+agent:
+	docker build -f deployments/docker/agent/Dockerfile . -t agent

README.md

@@ -81,6 +81,37 @@
 
     UI： http://127.0.0.1:8080/admin/ui
 
+- agent
+
+    agent 负责实时获取指定 应用/集群 下的所有配置文件到指定文件夹中
+
+    ```text
+    NAME:
+       agent - XConf agent client
+
+    USAGE:
+       main [global options] command [command options] [arguments...]
+
+    VERSION:
+       0.0.1
+
+    COMMANDS:
+         help, h  Shows a list of commands or help for one command
+
+    GLOBAL OPTIONS:
+       --url value, -u value      base url (default: "127.0.0.1:8080") [$XCONF_BASE_URL]
+       --app value, -a value      app name [$XCONF_APP_NAME]
+       --cluster value, -c value  cluster name [$XCONF_CLUSTER_NAME]
+       --dir value, -d value      directory (default: "/tmp") [$XCONF_DIR]
+       --help, -h                 show help
+       --version, -v              print the version
+
+    ```
+
+    ```shell script
+    docker run --name agent -v /tmp/docker-xconf:/tmp  -d  xuxu123/agent  -u http://xconf.mogutou.xyz -a test -c dev -d /tmp
+    ```
+
 ## 客户端
 
 - Golang  [client/example](client/example/main.go)

agent/Makefile

@@ -1,3 +1,3 @@
 
 run:
-	go run main.go
+	go run main.go  -u http://xconf.mogutou.xyz -a test -c dev -d /tmp/xconf

agent/config/config.go

@@ -0,0 +1,86 @@
+package config
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/micro-in-cn/XConf/agent/file"
+	"github.com/micro-in-cn/XConf/agent/source"
+	"github.com/micro/go-micro/util/log"
+)
+
+var ErrStopped = errors.New("config file stopped")
+
+type Config struct {
+	file    file.ConfigFile
+	source  source.Source
+	watcher source.Watcher
+	exit    chan interface{}
+	name    string
+}
+
+func New(filePatch string, url, appName, clusterName, namespaceName, format string) *Config {
+	return &Config{
+		file:   file.New(filePatch),
+		source: source.New(url, appName, clusterName, namespaceName),
+		exit:   make(chan interface{}),
+		name:   fmt.Sprintf("host:%s app:%s cluster:%s namespace:%s.%s", url, appName, clusterName, namespaceName, format),
+	}
+}
+
+func (s *Config) Init() error {
+	b, err := s.source.Read()
+	if err != nil {
+		return err
+	}
+
+	if err := s.file.Update(b); err != nil {
+		return err
+	}
+
+	s.watcher, err = s.source.Watch()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *Config) Sync() (err error) {
+	if s.watcher == nil {
+		return errors.New("Init function is not called ")
+	}
+
+	log.Info("Start watch ", s.name)
+	for {
+		select {
+		case <-s.exit:
+			return ErrStopped
+		default:
+			nb, err := s.watcher.Next()
+			if err != nil {
+				if err == source.ErrWatcherStopped {
+					return ErrStopped
+				}
+
+				log.Error("watch:", err)
+				continue
+			}
+
+			//log.Info("watch:\n", string(nb))
+			if err := s.file.Update(nb); err != nil {
+				log.Error("update:", err)
+				continue
+			}
+		}
+	}
+}
+
+func (s *Config) Stop() {
+	select {
+	case <-s.exit:
+	default:
+		_ = s.watcher.Stop()
+		close(s.exit)
+	}
+}

agent/file/file.go

@@ -0,0 +1,58 @@
+package file
+
+import (
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+
+	"github.com/micro/go-micro/util/log"
+)
+
+type ConfigFile interface {
+	Update(content []byte) error
+	Read() ([]byte, error)
+}
+
+type configFile struct {
+	filePath string
+}
+
+func New(filePatch string) ConfigFile {
+	return &configFile{
+		filePath: filePatch,
+	}
+}
+
+func (c *configFile) Update(content []byte) error {
+	if err := os.MkdirAll(filepath.Dir(c.filePath), 0755); err != nil {
+		return err
+	}
+	// create backup file
+	exist, err := ExistFile(c.filePath)
+	if err != nil {
+		return err
+	}
+	if exist {
+		if err := CopyFile(c.filePath, fmt.Sprintf("%s_backup", c.filePath)); err != nil {
+			return err
+		}
+	}
+
+	tmpFile := fmt.Sprintf("%s_tmp", c.filePath)
+	if err := ioutil.WriteFile(tmpFile, content, 0755); err != nil {
+		log.Error("write file error:", err)
+		return err
+	}
+
+	if err := os.Rename(tmpFile, c.filePath); err != nil {
+		log.Error("rename file error:", err)
+		return err
+	}
+
+	return nil
+}
+
+func (c *configFile) Read() ([]byte, error) {
+	return ioutil.ReadFile(c.filePath)
+}

agent/file/file_test.go

@@ -0,0 +1,25 @@
+package file
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+)
+
+func TestConfigFile(t *testing.T) {
+	file := New("/tmp/test.conf")
+
+	content := []byte(strings.Repeat("hello xconf agent\n", 512))
+	if err := file.Update(content); err != nil {
+		t.Fatal(err)
+	}
+
+	b, err := file.Read()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !bytes.Equal(content, b) {
+		t.Fatal()
+	}
+}

agent/file/util.go

@@ -0,0 +1,50 @@
+package file
+
+import (
+	"io"
+	"os"
+)
+
+func CopyFile(source string, dest string) error {
+	src, err := os.Open(source)
+	if err != nil {
+		return err
+	}
+	defer src.Close()
+
+	dst, err := os.Create(dest)
+	if err != nil {
+		return err
+	}
+	defer dst.Close()
+
+	_, err = io.Copy(dst, src)
+	if err != nil {
+		return err
+	}
+
+	info, err := os.Stat(source)
+	if err != nil {
+		err = os.Chmod(dest, info.Mode())
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func DeleteFile(file string) error {
+	return os.Remove(file)
+}
+
+func ExistFile(file string) (bool, error) {
+	_, err := os.Stat(file)
+	if err == nil {
+		return true, nil
+	}
+	if os.IsNotExist(err) {
+		return false, nil
+	}
+	return false, err
+}

agent/main.go

@@ -1,7 +1,85 @@
 package main
 
-import "fmt"
+import (
+	"errors"
+	"fmt"
+	"os"
+
+	"github.com/micro-in-cn/XConf/agent/server"
+	"github.com/micro/cli"
+)
 
 func main() {
-	fmt.Println("Hello XConf agent")
+	var (
+		baseURL     string
+		appName     string
+		clusterName string
+		dir         string
+	)
+
+	app := cli.NewApp()
+	app.Name = "agent"
+	app.Usage = "XConf agent client"
+	app.Version = "0.0.1"
+
+	app.Flags = []cli.Flag{
+		cli.StringFlag{
+			Name:        "url, u",
+			Value:       "127.0.0.1:8080",
+			Usage:       "base url",
+			EnvVar:      "XCONF_BASE_URL",
+			Destination: &baseURL,
+		},
+		cli.StringFlag{
+			Name:        "app, a",
+			Value:       "",
+			Usage:       "app name",
+			EnvVar:      "XCONF_APP_NAME",
+			Destination: &appName,
+		},
+		cli.StringFlag{
+			Name:        "cluster, c",
+			Value:       "",
+			Usage:       "cluster name",
+			EnvVar:      "XCONF_CLUSTER_NAME",
+			Destination: &clusterName,
+		},
+		cli.StringFlag{
+			Name:        "dir, d",
+			Value:       "/tmp",
+			Usage:       "directory",
+			EnvVar:      "XCONF_DIR",
+			Destination: &dir,
+		},
+	}
+
+	app.Action = func(c *cli.Context) error {
+		if len(baseURL) <= 0 {
+			return errors.New("base url cannot be empty")
+		}
+		if len(appName) <= 0 {
+			return errors.New("app name cannot be empty")
+		}
+		if len(clusterName) <= 0 {
+			return errors.New("cluster name cannot be empty")
+		}
+		if len(dir) <= 0 {
+			return errors.New("dir path cannot be empty")
+		}
+		return nil
+	}
+
+	if err := app.Run(os.Args); err != nil {
+		fmt.Println(err)
+		return
+	}
+
+	s := server.New(dir, baseURL, appName, clusterName)
+	if err := s.Init(); err != nil {
+		fmt.Println(err)
+		return
+	}
+
+	fmt.Println(s.Dir(), s.HostURL(), s.AppName(), s.ClusterName())
+	s.Run()
 }