opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-10-12 23:38:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #15908 from swordqiu/automated-cherry-pick-of-#15907-upstream-release-3.10

Browse Source 
Automated cherry pick of #15907: fix: lbagent containerize issues
This commit is contained in:
Zexi Li
2023-02-08 11:03:16 +08:00
committed by GitHub
parent fdfc330145 9ea346f38a
commit ee19e764ee
11 changed files with 223 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
build/docker/Dockerfile.lbagent
+8
View File
@@ -0,0 +1,8 @@
FROM registry.cn-beijing.aliyuncs.com/swordqiu/lbagent-base:v0.0.3
MAINTAINER "Yousong Zhou <zhouyousong@yunion.cn>"
ENV TZ UTC
RUN mkdir -p /opt/yunion/bin
ADD ./_output/alpine-build/bin/lbagent /opt/yunion/bin/lbagent
build/docker/Dockerfile.lbagent-base
+33
View File
@@ -0,0 +1,33 @@
FROM registry.cn-beijing.aliyuncs.com/yunionio/openvswitch:2.10.5-1
MAINTAINER "Jian QIU <qiujian@yunion.cn>"
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/*
pkg/compute/models/loadbalanceragents.go
+1
View File
@@ -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 {
pkg/hostman/guestfs/fsdriver/base.go
+45
View File
@@ -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
}
pkg/hostman/hostinfo/hostconsts/hostconsts.go
+2
View File
@@ -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"
pkg/hostman/hostinfo/hostinfo.go
+4 -38
View File
@@ -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()
}
pkg/hostman/system_service/telegraf.go
+21 -29
View File
@@ -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
}
pkg/lbagent/api.go
+11
View File
@@ -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()
pkg/lbagent/haproxy.go
+89 -3
View File
@@ -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()
pkg/lbagent/options.go
+1 -1
View File
@@ -24,7 +24,7 @@ import (
)
type LbagentCommonOptions struct {
common_options.CommonOptions
common_options.HostCommonOptions
// ApiLbagentId string `require:"true"`
ApiLbagentHbInterval int `default:"10"`
pkg/lbagent/service.go
+8 -1
View File
@@ -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)
}