feature: vpcagent sync handler

cmd/climc/shell/misc/vpcagent.go

@@ -0,0 +1,32 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package misc
+
+import (
+	"yunion.io/x/onecloud/pkg/mcclient"
+	"yunion.io/x/onecloud/pkg/mcclient/modules/vpcagent"
+)
+
+func init() {
+	type VpcAgentSyncOptions struct {
+	}
+	R(&VpcAgentSyncOptions{}, "vpcagent-sync", "Invoke sync of vpcagent", func(s *mcclient.ClientSession, args *VpcAgentSyncOptions) error {
+		err := vpcagent.VpcAgent.DoSync(s)
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+}

cmd/vpcagent/main.go

@@ -16,63 +16,17 @@ package main
 
 import (
 	"context"
-	"os"
-	"os/signal"
-	"sync"
-	"syscall"
 
-	"yunion.io/x/log"
-
-	"yunion.io/x/onecloud/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/atexit"
 	"yunion.io/x/onecloud/pkg/util/procutils"
-	"yunion.io/x/onecloud/pkg/vpcagent/options"
 	_ "yunion.io/x/onecloud/pkg/vpcagent/ovn"
-	"yunion.io/x/onecloud/pkg/vpcagent/worker"
+	"yunion.io/x/onecloud/pkg/vpcagent/service"
 )
 
 func main() {
 	defer atexit.Handle()
 
-	opts := &options.Options{}
-	commonOpts := &opts.CommonOptions
-	{
-		common_options.ParseOptions(opts, os.Args, "vpcagent.conf", "vpcagent")
-		app_common.InitAuth(commonOpts, func() {
-			log.Infof("auth finished ok")
-		})
-	}
-	if err := opts.ValidateThenInit(); err != nil {
-		log.Fatalf("opts validate: %s", err)
-	}
+	go procutils.WaitZombieLoop(context.TODO())
 
-	w := worker.NewWorker(opts)
-	if w == nil {
-		log.Fatalf("new worker failed")
-	}
-
-	{
-
-		ctx := context.Background()
-		ctx, cancelFunc := context.WithCancel(ctx)
-		go procutils.WaitZombieLoop(ctx)
-
-		wg := &sync.WaitGroup{}
-		ctx = context.WithValue(ctx, "wg", wg)
-		ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_APPNAME, "vpcagent")
-		wg.Add(1)
-		go w.Start(ctx)
-
-		go func() {
-			sigChan := make(chan os.Signal)
-			signal.Notify(sigChan, syscall.SIGINT)
-			signal.Notify(sigChan, syscall.SIGTERM)
-			sig := <-sigChan
-			log.Infof("signal received: %s", sig)
-			cancelFunc()
-		}()
-		wg.Wait()
-	}
+	service.StartService()
 }

pkg/apihelper/apihelper.go

@@ -16,18 +16,24 @@ package apihelper
 
 import (
 	"context"
+	"net/http"
 	"sync"
 	"time"
 
 	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
 
+	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/auth"
+	"yunion.io/x/onecloud/pkg/util/httputils"
 )
 
 const (
 	ErrSync = errors.Error("sync error")
+
+	MinSyncIntervalSeconds  = 10
+	MinRunDelayMilliseconds = 100
 )
 
 type APIHelper struct {
@@ -36,6 +42,8 @@ type APIHelper struct {
 	modelSetsCh chan IModelSets
 
 	mcclientSession *mcclient.ClientSession
+
+	tick *time.Timer
 }
 
 func NewAPIHelper(opts *Options, modelSets IModelSets) (*APIHelper, error) {
@@ -48,30 +56,72 @@ func NewAPIHelper(opts *Options, modelSets IModelSets) (*APIHelper, error) {
 	return helper, nil
 }
 
-func (h *APIHelper) Start(ctx context.Context) {
+func (h *APIHelper) getSyncInterval() time.Duration {
+	intv := h.opts.SyncIntervalSeconds
+	if intv < MinSyncIntervalSeconds {
+		intv = MinSyncIntervalSeconds
+	}
+	return time.Duration(intv) * time.Second
+}
+
+func (h *APIHelper) getRunDelay() time.Duration {
+	delay := h.opts.RunDelayMilliseconds
+	if delay < MinRunDelayMilliseconds {
+		delay = MinRunDelayMilliseconds
+	}
+	return time.Duration(delay) * time.Millisecond
+}
+
+func (h *APIHelper) addSyncHandler(app *appsrv.Application, prefix string) {
+	path := httputils.JoinPath(prefix, "sync")
+	app.AddHandler("POST", path, h.handlerSync)
+}
+
+func (h *APIHelper) handlerSync(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	h.scheduleSync()
+}
+
+func (h *APIHelper) Start(ctx context.Context, app *appsrv.Application, prefix string) {
 	defer func() {
 		log.Infoln("apihelper: bye")
 		wg := ctx.Value("wg").(*sync.WaitGroup)
 		wg.Done()
 	}()
 
+	if app != nil {
+		h.addSyncHandler(app, prefix)
+	}
+
 	h.run(ctx)
 
-	tickDuration := time.Duration(h.opts.SyncInterval) * time.Second
-	tick := time.NewTimer(tickDuration)
-	defer tick.Stop()
+	tickDuration := h.getSyncInterval()
+	h.tick = time.NewTimer(tickDuration)
+	defer func() {
+		tick := h.tick
+		h.tick = nil
+		tick.Stop()
+	}()
 
 	for {
 		select {
-		case <-tick.C:
+		case <-h.tick.C:
 			h.run(ctx)
-			tick.Reset(tickDuration)
+			h.tick.Reset(tickDuration)
 		case <-ctx.Done():
 			return
 		}
 	}
 }
 
+func (h *APIHelper) scheduleSync() {
+	if h.tick != nil {
+		if !h.tick.Stop() {
+			<-h.tick.C
+		}
+		h.tick.Reset(h.getRunDelay())
+	}
+}
+
 func (h *APIHelper) ModelSets() <-chan IModelSets {
 	return h.modelSetsCh
 }

pkg/apihelper/options.go

@@ -21,7 +21,8 @@ import (
 type Options struct {
 	common_options.CommonOptions
 
-	SyncInterval         int
+	SyncIntervalSeconds  int
+	RunDelayMilliseconds int
 	ListBatchSize        int
 	IncludeDetails       bool
 	IncludeOtherCloudEnv bool

pkg/apis/const.go

@@ -34,6 +34,7 @@ const (
 	SERVICE_TYPE_LOG               = "log"
 	SERVICE_TYPE_REGION            = "compute"
 	SERVICE_TYPE_CLOUDMON          = "cloudmon"
+	SERVICE_TYPE_VPCAGENT          = "vpcagent"
 
 	SERVICE_TYPE_ETCD     = "etcd"
 	SERVICE_TYPE_INFLUXDB = "influxdb"

pkg/cloudproxy/agent/options/options.go

@@ -24,7 +24,8 @@ type Options struct {
 	ProxyAgentInitWait         string `help:"duration to try and wait for init" default:"15s"`
 	proxyAgentInitWaitDuration time.Duration
 
-	APISyncInterval  int `default:"10"`
+	APISyncIntervalSeconds int `default:"10"`
+
 	APIListBatchSize int `default:"1024"`
 }
 
@@ -46,8 +47,8 @@ func (opts *Options) ValidateThenInit() error {
 	if opts.APIListBatchSize <= 20 {
 		opts.APIListBatchSize = 20
 	}
-	if opts.APISyncInterval <= 10 {
-		opts.APISyncInterval = 10
+	if opts.APISyncIntervalSeconds <= 10 {
+		opts.APISyncIntervalSeconds = 10
 	}
 
 	return nil

pkg/cloudproxy/agent/worker/worker.go

@@ -54,9 +54,9 @@ type Worker struct {
 func NewWorker(commonOpts *common_options.CommonOptions, opts *agentoptions.Options) *Worker {
 	modelSets := agentmodels.NewModelSets()
 	apiOpts := &apihelper.Options{
-		CommonOptions: *commonOpts,
-		SyncInterval:  opts.APISyncInterval,
-		ListBatchSize: opts.APIListBatchSize,
+		CommonOptions:       *commonOpts,
+		SyncIntervalSeconds: opts.APISyncIntervalSeconds,
+		ListBatchSize:       opts.APIListBatchSize,
 	}
 	apih, err := apihelper.NewAPIHelper(apiOpts, modelSets)
 	if err != nil {
@@ -177,7 +177,7 @@ func (w *Worker) Start(ctx context.Context) {
 		log.Errorf("init proxy agent: %v", err)
 		return
 	}
-	go w.apih.Start(ctx)
+	go w.apih.Start(ctx, nil, "")
 
 	const tickDur = 11 * time.Second
 	var (

pkg/lbagent/api.go

@@ -57,9 +57,10 @@ type ApiHelper struct {
 func NewApiHelper(opts *Options) (*ApiHelper, error) {
 	corpus := agentmodels.NewEmptyLoadbalancerCorpus()
 	apiOpts := &apihelper.Options{
-		CommonOptions: opts.CommonOptions,
-		SyncInterval:  opts.ApiSyncInterval,
-		ListBatchSize: opts.ApiListBatchSize,
+		CommonOptions:        opts.CommonOptions,
+		SyncIntervalSeconds:  opts.ApiSyncIntervalSeconds,
+		RunDelayMilliseconds: opts.ApiRunDelayMilliseconds,
+		ListBatchSize:        opts.ApiListBatchSize,
 	}
 	apih, err := apihelper.NewAPIHelper(apiOpts, corpus.ModelSets)
 	if err != nil {
@@ -87,10 +88,10 @@ func (h *ApiHelper) Run(ctx context.Context) {
 	defer h.stopOvnWorker()
 
 	wg.Add(1)
-	go h.apih.Start(ctx)
+	go h.apih.Start(ctx, nil, "")
 
 	hbTicker := time.NewTicker(time.Duration(h.opts.ApiLbagentHbInterval) * time.Second)
-	agentParamsSyncTicker := time.NewTicker(time.Duration(h.opts.ApiSyncInterval) * time.Second)
+	agentParamsSyncTicker := time.NewTicker(time.Duration(h.opts.ApiSyncIntervalSeconds) * time.Second)
 	defer hbTicker.Stop()
 	defer agentParamsSyncTicker.Stop()
 

pkg/lbagent/options.go

@@ -28,7 +28,9 @@ type LbagentOptions struct {
 	ApiLbagentHbInterval          int    `default:"10"`
 	ApiLbagentHbTimeoutRelaxation int    `default:"120" help:"If agent is to stale out in specified seconds in the future, consider it staled to avoid race condition when doing incremental api data fetch"`
 
-	ApiSyncInterval  int
+	ApiSyncIntervalSeconds  int `default:"10"`
+	ApiRunDelayMilliseconds int `default:"10"`
+
 	ApiListBatchSize int `default:"1024"`
 
 	DataPreserveN int `default:"8" help:"number of recent data to preserve on disk"`

pkg/mcclient/modulebase/base.go

@@ -19,9 +19,11 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"regexp"
 	"strings"
 
 	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
 
 	"yunion.io/x/onecloud/pkg/httperrors"
@@ -117,8 +119,17 @@ func (this *BaseManager) rawBaseUrlRequest(s *mcclient.ClientSession,
 	header http.Header, body io.Reader) (*http.Response, error) {
 	baseUrlF := func(baseurl string) string {
 		obj, _ := url.Parse(baseurl)
-		obj.Path = ""
-		return obj.String()
+		lastSlashPos := strings.LastIndex(obj.Path, "/")
+		if lastSlashPos >= 0 {
+			lastSeg := obj.Path[lastSlashPos+1:]
+			verReg := regexp.MustCompile(`^v\d+`)
+			if verReg.MatchString(lastSeg) {
+				obj.Path = obj.Path[:lastSlashPos]
+			}
+		}
+		ret := obj.String()
+		log.Debugf("baseurl %s ret %s", baseurl, ret)
+		return ret
 	}
 	return s.RawBaseUrlRequest(
 		this.serviceType, this.endpointType,

pkg/mcclient/modules/vpcagent/doc.go

@@ -0,0 +1,15 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package vpcagent // import "yunion.io/x/onecloud/pkg/mcclient/modules/vpcagent"

pkg/mcclient/modules/vpcagent/mod_vpcagent.go

@@ -0,0 +1,52 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package vpcagent
+
+import (
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/pkg/errors"
+
+	"yunion.io/x/onecloud/pkg/apis"
+	"yunion.io/x/onecloud/pkg/mcclient"
+	"yunion.io/x/onecloud/pkg/mcclient/modulebase"
+	"yunion.io/x/onecloud/pkg/mcclient/modules"
+)
+
+var (
+	VpcAgent VpcagentManager
+)
+
+func NewVpcagentManager(keyword, keywordPlural string, columns, adminColumns []string) modulebase.ResourceManager {
+	return modulebase.ResourceManager{
+		BaseManager: *modulebase.NewBaseManager(apis.SERVICE_TYPE_VPCAGENT, "", "", columns, adminColumns),
+		Keyword:     keyword, KeywordPlural: keywordPlural}
+}
+
+func init() {
+	VpcAgent = VpcagentManager{
+		ResourceManager: NewVpcagentManager("vpcagent", "vpcagent", []string{}, []string{}),
+	}
+	modules.Register(&VpcAgent)
+}
+
+type VpcagentManager struct {
+	modulebase.ResourceManager
+}
+
+func (agent *VpcagentManager) DoSync(s *mcclient.ClientSession) error {
+	params := jsonutils.NewDict()
+	_, err := agent.PerformAction(s, "api", "sync", params)
+	return errors.Wrap(err, "syncs")
+}

pkg/monitor/options/options.go

@@ -31,8 +31,9 @@ type AlerterOptions struct {
 	InitAlertResourceAdminRoleUsersIntervalSeconds int   `help:"internal to init alert resource admin role users " default:"3600"`
 	MonitorResourceSyncIntervalSeconds             int   `help:"internal to sync monitor resource,unit: h " default:"1"`
 
-	APISyncInterval  int `default:"3600"`
-	APIListBatchSize int `default:"1024"`
+	APISyncIntervalSeconds  int `default:"3600"`
+	APIRunDelayMilliseconds int `default:"5000"`
+	APIListBatchSize        int `default:"1024"`
 
 	WorkerCheckInterval int `default:"180"`
 

pkg/monitor/service/service.go

@@ -84,7 +84,7 @@ func StartService() {
 	if err != nil {
 		log.Fatalf("new worker failed: %v", err)
 	}
-	go worker.Start(context.Background())
+	go worker.Start(context.Background(), app, "")
 
 	//common_app.ServeForever(app, baseOpts)
 	InitInfluxDBSubscriptionHandlers(app, baseOpts)

pkg/monitor/worker/worker.go

@@ -23,6 +23,7 @@ import (
 	"yunion.io/x/pkg/errors"
 
 	"yunion.io/x/onecloud/pkg/apihelper"
+	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/monitor/models"
 	"yunion.io/x/onecloud/pkg/monitor/options"
 	"yunion.io/x/onecloud/pkg/vpcagent/worker"
@@ -39,7 +40,8 @@ func NewWorker(opts *options.AlerterOptions) (worker.IWorker, error) {
 	modelSets := man.GetModelSets()
 	apiOpts := &apihelper.Options{
 		CommonOptions:        opts.CommonOptions,
-		SyncInterval:         opts.APISyncInterval,
+		SyncIntervalSeconds:  opts.APISyncIntervalSeconds,
+		RunDelayMilliseconds: opts.APIRunDelayMilliseconds,
 		ListBatchSize:        opts.APIListBatchSize,
 		IncludeDetails:       true,
 		IncludeOtherCloudEnv: true,
@@ -56,13 +58,13 @@ func NewWorker(opts *options.AlerterOptions) (worker.IWorker, error) {
 	return w, nil
 }
 
-func (w *Worker) Start(ctx context.Context) {
+func (w *Worker) Start(ctx context.Context, app *appsrv.Application, prefix string) {
 	defer func() {
 		log.Infoln("monitor resource: worker bye")
 	}()
 
 	log.Infoln("start to get api Resource")
-	go w.apih.Start(ctx)
+	go w.apih.Start(ctx, nil, "")
 
 	var mss *models.MonitorResModelSets
 	for {

pkg/vpcagent/options/options.go

@@ -34,8 +34,9 @@ const (
 type VpcAgentOptions struct {
 	VpcProvider string `default:"ovn"`
 
-	APISyncInterval  int `default:"10"`
-	APIListBatchSize int `default:"1024"`
+	APISyncIntervalSeconds  int `default:"10"`
+	APIRunDelayMilliseconds int `default:"100"`
+	APIListBatchSize        int `default:"1024"`
 
 	OvnWorkerCheckInterval int    `default:"180"`
 	OvnNorthDatabase       string `help:"address for accessing ovn north database.  Default to local unix socket"`
@@ -60,9 +61,6 @@ func (opts *Options) ValidateThenInit() error {
 	if opts.APIListBatchSize <= 20 {
 		opts.APIListBatchSize = 20
 	}
-	if opts.APISyncInterval <= 10 {
-		opts.APISyncInterval = 10
-	}
 
 	if opts.OvnWorkerCheckInterval <= 60 {
 		opts.OvnWorkerCheckInterval = 60

pkg/vpcagent/ovn/worker.go

@@ -26,8 +26,10 @@ import (
 
 	"yunion.io/x/onecloud/pkg/apihelper"
 	apis "yunion.io/x/onecloud/pkg/apis/compute"
+	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/mcclient/auth"
 	mcclient_modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute"
+	"yunion.io/x/onecloud/pkg/util/httputils"
 	agentmodels "yunion.io/x/onecloud/pkg/vpcagent/models"
 	"yunion.io/x/onecloud/pkg/vpcagent/options"
 	"yunion.io/x/onecloud/pkg/vpcagent/ovnutil"
@@ -43,10 +45,11 @@ type Worker struct {
 func NewWorker(opts *options.Options) worker.IWorker {
 	modelSets := agentmodels.NewModelSets()
 	apiOpts := &apihelper.Options{
-		CommonOptions:  opts.CommonOptions,
-		SyncInterval:   opts.APISyncInterval,
-		ListBatchSize:  opts.APIListBatchSize,
-		IncludeDetails: false,
+		CommonOptions:        opts.CommonOptions,
+		SyncIntervalSeconds:  opts.APISyncIntervalSeconds,
+		RunDelayMilliseconds: opts.APIRunDelayMilliseconds,
+		ListBatchSize:        opts.APIListBatchSize,
+		IncludeDetails:       false,
 
 		IncludeOtherCloudEnv: false,
 	}
@@ -61,7 +64,7 @@ func NewWorker(opts *options.Options) worker.IWorker {
 	return w
 }
 
-func (w *Worker) Start(ctx context.Context) {
+func (w *Worker) Start(ctx context.Context, app *appsrv.Application, prefix string) {
 	wg := ctx.Value("wg").(*sync.WaitGroup)
 	defer func() {
 		log.Infoln("ovn: worker bye")
@@ -69,7 +72,7 @@ func (w *Worker) Start(ctx context.Context) {
 	}()
 
 	wg.Add(1)
-	go w.apih.Start(ctx)
+	go w.apih.Start(ctx, app, httputils.JoinPath(prefix, "api"))
 
 	tickDuration := time.Duration(w.opts.OvnWorkerCheckInterval) * time.Second
 	tick := time.NewTimer(tickDuration)

pkg/vpcagent/service/doc.go

@@ -0,0 +1,15 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service // import "yunion.io/x/onecloud/pkg/vpcagent/service"

pkg/vpcagent/service/service.go

@@ -0,0 +1,75 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+
+	"yunion.io/x/log"
+
+	"yunion.io/x/onecloud/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/vpcagent/options"
+	"yunion.io/x/onecloud/pkg/vpcagent/worker"
+)
+
+func StartService() {
+	opts := &options.Options{}
+	commonOpts := &opts.CommonOptions
+	{
+		common_options.ParseOptions(opts, os.Args, "vpcagent.conf", "vpcagent")
+		app_common.InitAuth(commonOpts, func() {
+			log.Infof("auth finished ok")
+		})
+	}
+	if err := opts.ValidateThenInit(); err != nil {
+		log.Fatalf("opts validate: %s", err)
+	}
+
+	app := app_common.InitApp(&opts.BaseOptions, false)
+
+	w := worker.NewWorker(opts)
+	if w == nil {
+		log.Fatalf("new worker failed")
+	}
+
+	go func() {
+		ctx := context.Background()
+		ctx, cancelFunc := context.WithCancel(ctx)
+
+		wg := &sync.WaitGroup{}
+		ctx = context.WithValue(ctx, "wg", wg)
+		ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_APPNAME, "vpcagent")
+		wg.Add(1)
+		go w.Start(ctx, app, "vpcagent")
+
+		go func() {
+			sigChan := make(chan os.Signal)
+			signal.Notify(sigChan, syscall.SIGINT)
+			signal.Notify(sigChan, syscall.SIGTERM)
+			sig := <-sigChan
+			log.Infof("signal received: %s", sig)
+			cancelFunc()
+		}()
+		wg.Wait()
+	}()
+
+	app_common.ServeForeverWithCleanup(app, &opts.BaseOptions, nil)
+}

pkg/vpcagent/worker/worker.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 
+	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/vpcagent/options"
 )
 
@@ -26,7 +27,7 @@ var workers = map[string]NewWorkerFunc{}
 type NewWorkerFunc func(opts *options.Options) IWorker
 
 type IWorker interface {
-	Start(ctx context.Context)
+	Start(ctx context.Context, app *appsrv.Application, prefix string)
 }
 
 func NewWorker(opts *options.Options) IWorker {