From 9ea346f38a89ba2acd1c859a5159a0eb78a7b4ae Mon Sep 17 00:00:00 2001 From: Qiu Jian Date: Wed, 8 Feb 2023 10:49:05 +0800 Subject: [PATCH] fix: lbagent containerize issues --- build/docker/Dockerfile.lbagent | 8 ++ build/docker/Dockerfile.lbagent-base | 33 +++++++ pkg/compute/models/loadbalanceragents.go | 1 + pkg/hostman/guestfs/fsdriver/base.go | 45 +++++++++ pkg/hostman/hostinfo/hostconsts/hostconsts.go | 2 + pkg/hostman/hostinfo/hostinfo.go | 42 +-------- pkg/hostman/system_service/telegraf.go | 50 +++++----- pkg/lbagent/api.go | 11 +++ pkg/lbagent/haproxy.go | 92 ++++++++++++++++++- pkg/lbagent/options.go | 2 +- pkg/lbagent/service.go | 9 +- 11 files changed, 223 insertions(+), 72 deletions(-) create mode 100644 build/docker/Dockerfile.lbagent create mode 100644 build/docker/Dockerfile.lbagent-base diff --git a/build/docker/Dockerfile.lbagent b/build/docker/Dockerfile.lbagent new file mode 100644 index 0000000000..04e77f8436 --- /dev/null +++ b/build/docker/Dockerfile.lbagent @@ -0,0 +1,8 @@ +FROM registry.cn-beijing.aliyuncs.com/swordqiu/lbagent-base:v0.0.3 + +MAINTAINER "Yousong Zhou " + +ENV TZ UTC + +RUN mkdir -p /opt/yunion/bin +ADD ./_output/alpine-build/bin/lbagent /opt/yunion/bin/lbagent diff --git a/build/docker/Dockerfile.lbagent-base b/build/docker/Dockerfile.lbagent-base new file mode 100644 index 0000000000..2196d20d45 --- /dev/null +++ b/build/docker/Dockerfile.lbagent-base @@ -0,0 +1,33 @@ +FROM registry.cn-beijing.aliyuncs.com/yunionio/openvswitch:2.10.5-1 + +MAINTAINER "Jian QIU " + +ENV TZ UTC + +ENV GOPATH=/root/go + +RUN sed -i 's!https://dl-cdn.alpinelinux.org/!https://mirrors.ustc.edu.cn/!g' /etc/apk/repositories + +RUN set -x \ + && apk update \ + && apk upgrade \ + && apk add --no-cache bash haproxy keepalived iptables \ + && apk add --no-cache --virtual .build-deps \ + build-base \ + make \ + go \ + git \ + gcc \ + libc-dev \ + libgcc \ + linux-headers \ + && mkdir -p $GOPATH/src/github.com/yyyar $GOPATH/bin \ + && cd $GOPATH/src/github.com/yyyar \ + && git clone --depth 1 https://github.com/yyyar/gobetween.git \ + && cd gobetween \ + && make deps \ + && make build \ + && cp bin/gobetween /bin/gobetween \ + && rm -fr $GOPATH \ + && apk del .build-deps \ + && rm -rf /var/cache/apk/* diff --git a/pkg/compute/models/loadbalanceragents.go b/pkg/compute/models/loadbalanceragents.go index 98475849b5..43a6b04083 100644 --- a/pkg/compute/models/loadbalanceragents.go +++ b/pkg/compute/models/loadbalanceragents.go @@ -784,6 +784,7 @@ func (lbagent *SLoadbalancerAgent) PerformLeaveCluster( oldClusterId := lbagent.ClusterId _, err := db.Update(lbagent, func() error { lbagent.ClusterId = "" + lbagent.HaState = api.LB_HA_STATE_UNKNOWN return nil }) if err != nil { diff --git a/pkg/hostman/guestfs/fsdriver/base.go b/pkg/hostman/guestfs/fsdriver/base.go index 0b58557a62..3008f3d01a 100644 --- a/pkg/hostman/guestfs/fsdriver/base.go +++ b/pkg/hostman/guestfs/fsdriver/base.go @@ -27,6 +27,10 @@ import ( "yunion.io/x/onecloud/pkg/cloudcommon/types" deployapi "yunion.io/x/onecloud/pkg/hostman/hostdeployer/apis" + "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/mcclient" + modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute" + "yunion.io/x/onecloud/pkg/util/procutils" ) type sGuestRootFsDriver struct { @@ -221,3 +225,44 @@ func MergeAuthorizedKeys(oldKeys string, pubkeys *deployapi.SSHKeys) string { } return strings.Join(keys, "\n") + "\n" } + +func DeployAdminAuthorizedKeys(s *mcclient.ClientSession) error { + sshDir := path.Join("/root", ".ssh") + output, err := procutils.NewRemoteCommandAsFarAsPossible("mkdir", "-p", sshDir).Output() + if err != nil { + return errors.Wrapf(err, "mkdir .ssh %s", output) + } + + query := jsonutils.NewDict() + query.Set("admin", jsonutils.JSONTrue) + ret, err := modules.Sshkeypairs.List(s, query) + if err != nil { + return errors.Wrap(err, "modules.Sshkeypairs.List") + } + if len(ret.Data) == 0 { + return errors.Wrap(httperrors.ErrNotFound, "Not found admin sshkey") + } + keys := ret.Data[0] + adminPublicKey, _ := keys.GetString("public_key") + pubKeys := &deployapi.SSHKeys{AdminPublicKey: adminPublicKey} + + var oldKeys string + authFile := path.Join(sshDir, "authorized_keys") + if procutils.NewRemoteCommandAsFarAsPossible("test", "-f", authFile).Run() == nil { + output, err := procutils.NewRemoteCommandAsFarAsPossible("cat", authFile).Output() + if err != nil { + return errors.Wrapf(err, "cat: %s", output) + } + oldKeys = string(output) + } + newKeys := MergeAuthorizedKeys(oldKeys, pubKeys) + if output, err := procutils.NewRemoteCommandAsFarAsPossible( + "sh", "-c", fmt.Sprintf("echo '%s' > %s", newKeys, authFile)).Output(); err != nil { + return errors.Wrapf(err, "write public keys: %s", output) + } + if output, err := procutils.NewRemoteCommandAsFarAsPossible( + "chmod", "0644", authFile).Output(); err != nil { + return errors.Wrapf(err, "chmod failed %s", output) + } + return nil +} diff --git a/pkg/hostman/hostinfo/hostconsts/hostconsts.go b/pkg/hostman/hostinfo/hostconsts/hostconsts.go index c0e396cb92..98bc1a3d14 100644 --- a/pkg/hostman/hostinfo/hostconsts/hostconsts.go +++ b/pkg/hostman/hostinfo/hostconsts/hostconsts.go @@ -23,8 +23,10 @@ const ( TELEGRAF_TAG_ONECLOUD_BRAND = "OneCloud" TELEGRAF_TAG_ONECLOUD_RES_TYPE = "host" + TELEGRAF_TAG_ONECLOUD_RES_TYPE_LBAGENT = "lbagent" TELEGRAF_TAG_ONECLOUD_HOST_TYPE_HOST = "host" TELEGRAF_TAG_ONECLOUD_HOST_TYPE_CONTROLLER = "controller" + TELEGRAF_TAG_ONECLOUD_HOST_TYPE_LBAGENT = "lbagent" SHUTDOWN_SERVERS = "shutdown-servers" diff --git a/pkg/hostman/hostinfo/hostinfo.go b/pkg/hostman/hostinfo/hostinfo.go index 35ab8e6980..addfcc9a36 100644 --- a/pkg/hostman/hostinfo/hostinfo.go +++ b/pkg/hostman/hostinfo/hostinfo.go @@ -47,7 +47,6 @@ import ( "yunion.io/x/onecloud/pkg/cloudcommon/types" "yunion.io/x/onecloud/pkg/hostman/guestfs/fsdriver" "yunion.io/x/onecloud/pkg/hostman/host_health" - deployapi "yunion.io/x/onecloud/pkg/hostman/hostdeployer/apis" "yunion.io/x/onecloud/pkg/hostman/hostinfo/hostbridge" "yunion.io/x/onecloud/pkg/hostman/hostinfo/hostconsts" "yunion.io/x/onecloud/pkg/hostman/hostutils" @@ -1896,46 +1895,13 @@ func (h *SHostInfo) probeSyncIsolatedDevices() (*jsonutils.JSONArray, error) { func (h *SHostInfo) deployAdminAuthorizedKeys() { onErr := func(format string, args ...interface{}) { h.onFail(fmt.Sprintf(format, args...)) + } + err := fsdriver.DeployAdminAuthorizedKeys(h.GetSession()) + if err != nil { + onErr("DeployAdminAuthorizedKeys fails: %s", err) return } - sshDir := path.Join("/root", ".ssh") - output, err := procutils.NewRemoteCommandAsFarAsPossible("mkdir", "-p", sshDir).Output() - if err != nil { - onErr("mkdir .ssh failed %s %s", output, err) - } - - query := jsonutils.NewDict() - query.Set("admin", jsonutils.JSONTrue) - ret, err := modules.Sshkeypairs.List(h.GetSession(), query) - if err != nil { - onErr("Get admin sshkey: %v", err) - } - if len(ret.Data) == 0 { - onErr("Not found admin sshkey") - } - keys := ret.Data[0] - adminPublicKey, _ := keys.GetString("public_key") - pubKeys := &deployapi.SSHKeys{AdminPublicKey: adminPublicKey} - - var oldKeys string - authFile := path.Join(sshDir, "authorized_keys") - if procutils.NewRemoteCommandAsFarAsPossible("test", "-f", authFile).Run() == nil { - output, err := procutils.NewRemoteCommandAsFarAsPossible("cat", authFile).Output() - if err != nil { - onErr("cat auth file %s %s", output, err) - } - oldKeys = string(output) - } - newKeys := fsdriver.MergeAuthorizedKeys(oldKeys, pubKeys) - if output, err := procutils.NewRemoteCommandAsFarAsPossible( - "sh", "-c", fmt.Sprintf("echo '%s' > %s", newKeys, authFile)).Output(); err != nil { - onErr("write public keys: %s %s", output, err) - } - if output, err := procutils.NewRemoteCommandAsFarAsPossible( - "chmod", "0644", authFile).Output(); err != nil { - onErr("chmod failed %s %s", output, err) - } h.onSucc() } diff --git a/pkg/hostman/system_service/telegraf.go b/pkg/hostman/system_service/telegraf.go index 0f08f3ad29..e93fa3a790 100644 --- a/pkg/hostman/system_service/telegraf.go +++ b/pkg/hostman/system_service/telegraf.go @@ -48,7 +48,14 @@ func (s *STelegraf) GetConfig(kwargs map[string]interface{}) string { } conf += "\n" conf += "[agent]\n" - conf += " interval = \"60s\"\n" + intVal := 60 + if v, ok := kwargs["interval"]; ok { + intvalInt, _ := v.(int) + if intvalInt > 0 { + intVal = intvalInt + } + } + conf += fmt.Sprintf(" interval = \"%ds\"\n", intVal) conf += " round_interval = true\n" conf += " metric_batch_size = 1000\n" conf += " metric_buffer_limit = 10000\n" @@ -83,28 +90,6 @@ func (s *STelegraf) GetConfig(kwargs map[string]interface{}) string { conf += " timeout = \"30s\"\n" conf += "\n" } - if kafka, ok := kwargs["kafka"]; ok { - ka, _ := kafka.(map[string]interface{}) - bks, _ := ka["brokers"] - tbk, _ := bks.([]string) - brokers := []string{} - for _, b := range tbk { - brokers = append(brokers, fmt.Sprintf("\"%s\"", b[len("kafka://\n"):])) - } - conf += "[[outputs.kafka]]\n" - conf += fmt.Sprintf(" brokers = [%s]\n", strings.Join(brokers, ", ")) - - topic, _ := ka["topic"] - itopic, _ := topic.(string) - conf += fmt.Sprintf(" topic = \"%s\"\n", itopic) - conf += " compression_codec = 0\n" - conf += " required_acks = -1\n" - conf += " max_retry = 3\n" - conf += " data_format = \"json\"\n" - conf += " json_timestamp_units = \"1ms\"\n" - conf += " routing_tag = \"host\"\n" - conf += "\n" - } conf += "[[inputs.cpu]]\n" conf += " percpu = false\n" conf += " totalcpu = true\n" @@ -146,12 +131,9 @@ func (s *STelegraf) GetConfig(kwargs map[string]interface{}) string { conf += fmt.Sprintf(" interfaces = [%s]\n", strings.Join(infs, ", ")) conf += "\n" for _, n := range ns { - iname, _ := n["name"] - name, _ := iname.(string) - ialias, _ := n["alias"] - alias, _ := ialias.(string) - ispeed, _ := n["speed"] - speed, _ := ispeed.(int) + name, _ := n["name"].(string) + alias, _ := n["alias"].(string) + speed, _ := n["speed"].(int) conf += " [[inputs.net.interface_conf]]\n" conf += fmt.Sprintf(" name = \"%s\"\n", name) @@ -182,6 +164,16 @@ func (s *STelegraf) GetConfig(kwargs map[string]interface{}) string { conf += " data_source = \"body\"\n" conf += " data_format = \"influx\"\n" conf += "\n" + if haproxyConf, ok := kwargs["haproxy"]; ok { + haproxyConfMap, _ := haproxyConf.(map[string]interface{}) + haIntVal, _ := haproxyConfMap["interval"].(int) + statsSocket, _ := haproxyConfMap["stats_socket_path"].(string) + conf += "[[inputs.haproxy]]\n" + conf += fmt.Sprintf(" interval = \"%ds\"\n", haIntVal) + conf += fmt.Sprintf(" servers = [\"%s\"]\n", statsSocket) + conf += " keep_field_names = true\n" + conf += "\n" + } return conf } diff --git a/pkg/lbagent/api.go b/pkg/lbagent/api.go index 0633156c5d..c2e8889f00 100644 --- a/pkg/lbagent/api.go +++ b/pkg/lbagent/api.go @@ -29,6 +29,7 @@ import ( "yunion.io/x/onecloud/pkg/apihelper" api "yunion.io/x/onecloud/pkg/apis/compute" computemodels "yunion.io/x/onecloud/pkg/compute/models" + "yunion.io/x/onecloud/pkg/hostman/guestfs/fsdriver" agentmodels "yunion.io/x/onecloud/pkg/lbagent/models" agentutils "yunion.io/x/onecloud/pkg/lbagent/utils" "yunion.io/x/onecloud/pkg/mcclient" @@ -82,7 +83,17 @@ func NewApiHelper(opts *Options, lbagentId string) (*ApiHelper, error) { return helper, nil } +func (h *ApiHelper) deployAdminAuthorizedKeys(ctx context.Context) { + err := fsdriver.DeployAdminAuthorizedKeys(h.adminClientSession(ctx)) + if err != nil { + log.Errorf("DeployAdminAuthorizedKeys %s", err) + } +} + func (h *ApiHelper) Run(ctx context.Context) { + // deploy host admin key + h.deployAdminAuthorizedKeys(ctx) + wg := ctx.Value("wg").(*sync.WaitGroup) defer func() { wg.Done() diff --git a/pkg/lbagent/haproxy.go b/pkg/lbagent/haproxy.go index 71a658cf57..785d1804a5 100644 --- a/pkg/lbagent/haproxy.go +++ b/pkg/lbagent/haproxy.go @@ -22,6 +22,7 @@ import ( "os" "os/exec" "path/filepath" + "reflect" "strings" "sync" "syscall" @@ -29,20 +30,27 @@ import ( "yunion.io/x/log" "yunion.io/x/pkg/errors" + "yunion.io/x/pkg/gotypes" + "yunion.io/x/onecloud/pkg/hostman/hostinfo/hostconsts" + "yunion.io/x/onecloud/pkg/hostman/system_service" agentmodels "yunion.io/x/onecloud/pkg/lbagent/models" agentutils "yunion.io/x/onecloud/pkg/lbagent/utils" + "yunion.io/x/onecloud/pkg/util/sysutils" ) type HaproxyHelper struct { opts *Options + lbagentId string + configDirMan *agentutils.ConfigDirManager } -func NewHaproxyHelper(opts *Options) (*HaproxyHelper, error) { +func NewHaproxyHelper(opts *Options, lbagentId string) (*HaproxyHelper, error) { helper := &HaproxyHelper{ opts: opts, + lbagentId: lbagentId, configDirMan: agentutils.NewConfigDirManager(opts.haproxyConfigDir), } { @@ -56,6 +64,9 @@ func NewHaproxyHelper(opts *Options) (*HaproxyHelper, error) { return nil, fmt.Errorf("sysctl: %s", err) } } + + system_service.Init() + return helper, nil } @@ -189,7 +200,7 @@ func (h *HaproxyHelper) handleUseCorpusCmd(ctx context.Context, cmd *LbagentCmd) p := filepath.Join(dir, "telegraf.conf") err := ioutil.WriteFile(p, d, agentutils.FileModeFile) if err == nil { - err := h.reloadTelegraf(ctx) + err := h.reloadTelegraf(ctx, agentParams) if err != nil { log.Errorf("reloading telegraf.conf failed: %s", err) } @@ -411,7 +422,82 @@ func (h *HaproxyHelper) telegrafPidFile() *agentutils.PidFile { return pf } -func (h *HaproxyHelper) reloadTelegraf(ctx context.Context) error { +func (h *HaproxyHelper) reloadTelegraf(ctx context.Context, agentParams *agentmodels.AgentParams) error { + if h.opts.EnableRemoteExecutor { + return h.remoteReloadTelegraf(ctx, agentParams) + } else { + return h.localReloadTelegraf(ctx) + } +} + +func (h *HaproxyHelper) remoteReloadTelegraf(ctx context.Context, agentParams *agentmodels.AgentParams) error { + telegraf := system_service.GetService("telegraf") + conf := map[string]interface{}{} + conf["hostname"] = h.getHostname() + conf["tags"] = map[string]string{ + "id": h.lbagentId, + "lbagent_id": h.lbagentId, + "region": h.opts.Region, + "lbagent_ip": h.opts.AccessIp, + hostconsts.TELEGRAF_TAG_KEY_BRAND: hostconsts.TELEGRAF_TAG_ONECLOUD_BRAND, + hostconsts.TELEGRAF_TAG_KEY_RES_TYPE: hostconsts.TELEGRAF_TAG_ONECLOUD_RES_TYPE_LBAGENT, + hostconsts.TELEGRAF_TAG_KEY_HOST_TYPE: hostconsts.TELEGRAF_TAG_ONECLOUD_HOST_TYPE_LBAGENT, + } + conf["nics"] = h.getNicsTelegrafConf() + if len(agentParams.AgentModel.Params.Telegraf.InfluxDbOutputUrl) > 0 { + conf["influxdb"] = map[string]interface{}{ + "url": []string{ + agentParams.AgentModel.Params.Telegraf.InfluxDbOutputUrl, + }, + "database": agentParams.AgentModel.Params.Telegraf.InfluxDbOutputName, + } + } + conf["haproxy"] = map[string]interface{}{ + "interval": agentParams.AgentModel.Params.Telegraf.HaproxyInputInterval, + "stats_socket_path": h.haproxyStatsSocketFile(), + } + oldConf := telegraf.GetConf() + log.Debugf("old config: %s", oldConf) + log.Debugf("new config: %s", conf) + if gotypes.IsNil(oldConf) || !reflect.DeepEqual(oldConf, conf) { + log.Debugf("telegraf config: %s", conf) + telegraf.SetConf(conf) + telegraf.BgReloadConf(conf) + } + return nil +} + +func (h *HaproxyHelper) getNicsTelegrafConf() []map[string]interface{} { + var ret = make([]map[string]interface{}, 0) + phyNics, _ := sysutils.Nics() + for _, pnic := range phyNics { + ret = append(ret, map[string]interface{}{ + "name": pnic.Dev, + "speed": pnic.Speed, + }) + } + return ret +} + +func (h *HaproxyHelper) getHostname() string { + hn, err := os.Hostname() + if err != nil { + log.Fatalf("fail to get hostname %s", err) + return "" + } + dotIdx := strings.IndexByte(hn, '.') + if dotIdx >= 0 { + hn = hn[:dotIdx] + } + hn = strings.ToLower(hn) + if len(hn) == 0 { + hn = "host" + } + masterIp := h.opts.AccessIp + return hn + "-" + strings.Replace(masterIp, ".", "-", -1) +} + +func (h *HaproxyHelper) localReloadTelegraf(ctx context.Context) error { pidFile := h.telegrafPidFile() { proc, confirmed, err := pidFile.ConfirmOrUnlink() diff --git a/pkg/lbagent/options.go b/pkg/lbagent/options.go index 52880dd1b4..bc6d0c9dfa 100644 --- a/pkg/lbagent/options.go +++ b/pkg/lbagent/options.go @@ -24,7 +24,7 @@ import ( ) type LbagentCommonOptions struct { - common_options.CommonOptions + common_options.HostCommonOptions // ApiLbagentId string `require:"true"` ApiLbagentHbInterval int `default:"10"` diff --git a/pkg/lbagent/service.go b/pkg/lbagent/service.go index 1db38b498f..12785e0a65 100644 --- a/pkg/lbagent/service.go +++ b/pkg/lbagent/service.go @@ -21,12 +21,14 @@ import ( "sync" "syscall" + execlient "yunion.io/x/executor/client" "yunion.io/x/log" "yunion.io/x/pkg/appctx" app_common "yunion.io/x/onecloud/pkg/cloudcommon/app" common_options "yunion.io/x/onecloud/pkg/cloudcommon/options" "yunion.io/x/onecloud/pkg/util/fileutils2" + "yunion.io/x/onecloud/pkg/util/procutils" ) func StartService() { @@ -52,6 +54,11 @@ func StartService() { log.Fatalf("opts validate: %s", err) } + if opts.EnableRemoteExecutor { + execlient.Init(opts.ExecutorSocketPath) + procutils.SetRemoteExecutor() + } + // register lbagent ctx := context.Background() ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_APPNAME, "lbagent") @@ -70,7 +77,7 @@ func StartService() { } } { - haproxyHelper, err = NewHaproxyHelper(opts) + haproxyHelper, err = NewHaproxyHelper(opts, lbagentId) if err != nil { log.Fatalf("init haproxy helper failed: %s", err) }