bubblecopy/config.go at main · bubble-diff/bubblecopy · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"sync"
	"time"

	"github.com/sirupsen/logrus"
)

type config struct {
	// Taskid Diff任务ID
	Taskid int64 `json:"taskid"`
	// Secret 访问对应id任务配置的密钥
	Secret string `json:"secret"`
	// Device 网卡名称
	Device string `json:"interface"`
	// Port 被测服务端口
	Port string `json:"service_port"`
	// ReplaySvrAddr bubblereplay服务地址
	ReplaySvrAddr string `json:"replay_svr_addr"`

	// DeviceIPv4 网卡ipv4地址
	DeviceIPv4 string

	mu            sync.Mutex
	isTaskRunning bool
}

var configuration = config{}

func (c *config) init() {
	bytes, err := os.ReadFile(SettingsFilePath)
	if err != nil {
		logrus.Error(err)
		logrus.Fatal("Need settings.json to get configuration.")
	}
	err = json.Unmarshal(bytes, &configuration)
	if err != nil {
		logrus.Fatal(err)
	}

	if configuration.ReplaySvrAddr == "" {
		logrus.Fatal("bubblereplay server addr not set")
	}

	c.DeviceIPv4, err = getDeviceIpv4(c.Device)
	if err != nil {
		logrus.Error(err)
		logrus.Fatalf("%s: this device has no ipv4 address.", c.Device)
	}

	// 启动后台去执行一些自动更新的动作
	go func() {
		logrus.Info("configuration background started.")
		for {
			c.setIsTaskRunning()
			c.reportDepolyed()
			time.Sleep(5 * time.Second)
		}
	}()

	logrus.WithField(
		"configuration",
		fmt.Sprintf("%+v", configuration),
	).Debug("configuration initialized.")
}

func (c *config) getIsTaskRunning() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.isTaskRunning
}

func (c *config) setIsTaskRunning() {
	var err error
	var apiResp struct {
		Err       string `json:"err"`
		IsRunning bool   `json:"is_running"`
	}

	logrus.Infof("[setIsTaskRunning] updating TaskID=%d status...", c.Taskid)
	// call bubblereplay
	api := fmt.Sprintf("http://%s%s/%d", c.ReplaySvrAddr, ApiTaskStatus, c.Taskid)
	resp, err := http.Get(api)
	if err != nil {
		logrus.Errorf("[setIsTaskRunning] call api failed, %s", err)
		return
	}

	err = json.NewDecoder(resp.Body).Decode(&apiResp)
	if err != nil {
		logrus.Errorf("[setIsTaskRunning] decode json response failed, %s", err)
		return
	}
	resp.Body.Close()

	if apiResp.Err != "" {
		logrus.Errorf("[setIsTaskRunning] response return error, %s", err)
		return
	}

	// update state
	c.mu.Lock()
	defer c.mu.Unlock()
	c.isTaskRunning = apiResp.IsRunning
}

func (c *config) reportDepolyed() {
	var err error
	var apiBody = struct {
		TaskID int64 `json:"task_id"`
		// Addr 基准服务地址
		Addr string `json:"addr"`
	}{
		TaskID: c.Taskid,
		Addr:   fmt.Sprintf("%s:%s", c.DeviceIPv4, c.Port),
	}
	var apiResp struct {
		Err string `json:"err"`
	}

	logrus.Infof("[reportDepolyed] reporting TaskID=%d has been deployed the bubblecopy...", c.Taskid)
	// call bubblereplay
	api := fmt.Sprintf("http://%s%s", c.ReplaySvrAddr, ApiSetDeployed)
	data, err := json.Marshal(apiBody)
	resp, err := http.Post(api, "application/json", bytes.NewReader(data))
	if err != nil {
		logrus.Errorf("[reportDepolyed] call api failed, %s", err)
		return
	}

	err = json.NewDecoder(resp.Body).Decode(&apiResp)
	if err != nil {
		logrus.Errorf("[reportDepolyed] decode json response failed, %s", err)
		return
	}
	resp.Body.Close()

	if apiResp.Err != "" {
		logrus.Errorf("[reportDepolyed] response return error, %s", err)
		return
	}
}