Proposal: Wire for dependency injection

core.go

@@ -0,0 +1,163 @@
+//  Copyright (c) 2020 Cisco and/or its affiliates.
+//
+//  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 cninfra
+
+import (
+	"context"
+
+	"github.com/google/wire"
+
+	"go.ligato.io/cn-infra/v2/config"
+	"go.ligato.io/cn-infra/v2/db/keyval"
+	"go.ligato.io/cn-infra/v2/health/probe"
+	"go.ligato.io/cn-infra/v2/health/statuscheck"
+	"go.ligato.io/cn-infra/v2/logging"
+	"go.ligato.io/cn-infra/v2/logging/logmanager"
+	"go.ligato.io/cn-infra/v2/rpc/grpc"
+	"go.ligato.io/cn-infra/v2/rpc/prometheus"
+	"go.ligato.io/cn-infra/v2/rpc/rest"
+	"go.ligato.io/cn-infra/v2/servicelabel"
+)
+
+type Base struct {
+	ServiceLabel *servicelabel.Plugin // implements: servicelabel.ReaderAPI
+	StatusCheck  *statuscheck.Plugin  // implements: statuscheck.PluginStatusWriter, statuscheck.StatusReader
+
+}
+
+func (base Base) Start(ctx context.Context) error {
+	logging.Debugf("Base Start()")
+
+	if err := base.ServiceLabel.InitLabel(); err != nil {
+		return err
+	}
+	if err := base.StatusCheck.StartProbing(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+type Server struct {
+	HTTP *rest.Plugin // implements: rest.HTTPHandlers
+	GRPC *grpc.Plugin // implements: grpc.Server
+}
+
+func (server Server) Start(ctx context.Context) error {
+	logging.Debugf("Server Start()")
+
+	if err := server.HTTP.StartServing(); err != nil {
+		logging.Fatalf("HTTP serving failed: %v", err)
+	}
+	if err := server.GRPC.StartServing(); err != nil {
+		logging.Fatalf("GRPC serving failed: %v", err)
+	}
+
+	return nil
+}
+
+type KVStore struct {
+	keyval.KvProtoPlugin
+}
+
+func ProvideServiceLabelReaderAPI(core Base) servicelabel.ReaderAPI {
+	if core.ServiceLabel == nil {
+		return nil
+	}
+	return core.ServiceLabel
+}
+
+func ProvideStatusCheckStatusReader(core Base) statuscheck.StatusReader {
+	if core.StatusCheck == nil {
+		return nil
+	}
+	return core.StatusCheck
+}
+
+func ProvideStatusCheckPluginStatusWriter(core Base) statuscheck.PluginStatusWriter {
+	if core.StatusCheck == nil {
+		return nil
+	}
+	return core.StatusCheck
+}
+
+var CoreProviders = wire.NewSet(
+	ProvideServiceLabelReaderAPI,
+	ProvideStatusCheckStatusReader,
+	ProvideStatusCheckPluginStatusWriter,
+)
+
+func ProvideRestHTTPHandlers(server Server) rest.HTTPHandlers {
+	if server.HTTP == nil {
+		return nil
+	}
+	return server.HTTP
+}
+
+func ProvideGrpcServer(server Server) grpc.Server {
+	if server.GRPC == nil {
+		return nil
+	}
+	return server.GRPC
+}
+
+var ServerProviders = wire.NewSet(
+	ProvideRestHTTPHandlers,
+	ProvideGrpcServer,
+)
+
+var WireDefaultCore = wire.NewSet(
+	servicelabel.WireDefault,
+	statuscheck.WireDefault,
+)
+
+var WireDefaultServer = wire.NewSet(
+	rest.WireDefault,
+	grpc.WireDefault,
+)
+
+var WirePrometheusProbe = wire.NewSet(
+	probe.WireDefault,
+	prometheus.WireDefault,
+)
+
+var WireDefaultLogRegistry = wire.NewSet(
+	wire.InterfaceValue(new(logging.Registry), logging.DefaultRegistry),
+)
+
+var WireDefaultConfig = wire.NewSet(
+	wire.InterfaceValue(new(config.Config), config.DefaultConfig),
+)
+
+var WireLogManager = wire.NewSet(
+	WireDefaultLogRegistry,
+	logmanager.WireDefault,
+)
+
+/*func StatusCheckProvider(deps statuscheck.Deps, conf *statuscheck.Config) (*statuscheck.Plugin, func(), error) {
+	p := &statuscheck.Plugin{Deps: deps}
+	p.conf = conf
+	p.Log = logging.ForPlugin("status-check")
+	cancel := func() {
+		if err := p.Close(); err != nil {
+			p.Log.Error(err)
+		}
+	}
+	return p, cancel, p.Init()
+}
+
+var WireStatusCheck = wire.NewSet(
+	StatusCheckProvider,
+)*/

datasync/resync/plugin_impl_resync.go

@@ -48,6 +48,8 @@ type Deps struct {
 
 // Init initializes variables.
 func (p *Plugin) Init() error {
+	p.Log.Debug("Init()")
+
 	p.registrations = make(map[string]*registration)
 	return nil
 }

datasync/resync/wire.go

@@ -0,0 +1,37 @@
+//  Copyright (c) 2020 Cisco and/or its affiliates.
+//
+//  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 resync
+
+import (
+	"github.com/google/wire"
+
+	"go.ligato.io/cn-infra/v2/logging"
+)
+
+var WireDefault = wire.NewSet(
+	Provider,
+	wire.Bind(new(Subscriber), new(*Plugin)),
+)
+
+func Provider() (*Plugin, func(), error) {
+	p := &Plugin{}
+	p.Log = logging.ForPlugin("resync")
+	cancel := func() {
+		if err := p.Close(); err != nil {
+			p.Log.Error(err)
+		}
+	}
+	return p, cancel, p.Init()
+}

db/keyval/consul/plugin.go

@@ -130,14 +130,16 @@ func (p *Plugin) Disabled() bool {
 
 func (p *Plugin) getConfig() (*Config, error) {
 	var cfg Config
-	found, err := p.Cfg.LoadValue(&cfg)
-	if err != nil {
-		return nil, err
-	}
-	if !found {
-		p.Log.Info("Consul config not found, skip loading this plugin")
-		p.disabled = true
-		return nil, nil
+	if p.Cfg != nil {
+		found, err := p.Cfg.LoadValue(&cfg)
+		if err != nil {
+			return nil, err
+		}
+		if !found {
+			p.Log.Info("Consul config not found, skip loading this plugin")
+			p.disabled = true
+			return nil, nil
+		}
 	}
 	return &cfg, nil
 }

db/keyval/consul/wire.go

@@ -0,0 +1,46 @@
+//  Copyright (c) 2020 Cisco and/or its affiliates.
+//
+//  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 consul
+
+import (
+	"github.com/google/wire"
+
+	"go.ligato.io/cn-infra/v2/config"
+	"go.ligato.io/cn-infra/v2/db/keyval"
+	"go.ligato.io/cn-infra/v2/logging"
+)
+
+var WireDefault = wire.NewSet(
+	Provider,
+	ConfigProvider,
+	wire.Struct(new(Deps), "StatusCheck", "Resync"),
+	wire.Bind(new(keyval.KvProtoPlugin), new(*Plugin)),
+)
+
+func ConfigProvider(conf config.Config) *Config {
+	var cfg Config
+	if err := conf.UnmarshalKey("consul", &cfg); err != nil {
+		logging.Errorf("unmarshal key failed: %v", err)
+	}
+	return &cfg
+}
+
+func Provider(deps Deps, cfg *Config) (keyval.KvProtoPlugin, func(), error) {
+	p := &Plugin{}
+	p.Deps = deps
+	p.Config = cfg
+	p.Log = logging.ForPlugin("consul-client")
+	return p, func() {}, nil
+}

db/keyval/etcd/config.go

@@ -66,6 +66,8 @@ type ClientConfig struct {
 }
 
 const (
+	defaultEndpoint = "127.0.0.1:2379"
+
 	// defaultDialTimeout defines the default timeout for connecting to etcd.
 	defaultDialTimeout = 1 * time.Second
 
@@ -76,6 +78,15 @@ const (
 	defaultSessionTTL = 5
 )
 
+func DefaultConfig() *Config {
+	return &Config{
+		DialTimeout: defaultDialTimeout,
+		OpTimeout:   defaultOpTimeout,
+		SessionTTL:  defaultSessionTTL,
+		Endpoints:   []string{defaultEndpoint},
+	}
+}
+
 // ConfigToClient transforms yaml configuration <yc> modelled by Config
 // into ClientConfig, which is ready for use with the underlying coreos/etcd
 // package.
@@ -113,7 +124,7 @@ func ConfigToClient(yc *Config) (*ClientConfig, error) {
 		} else if ep := os.Getenv("ETCDV3_ENDPOINTS"); ep != "" { // this provides backwards compatiblity
 			cfg.Endpoints = strings.Split(ep, ",")
 		} else {
-			cfg.Endpoints = []string{"127.0.0.1:2379"}
+			cfg.Endpoints = []string{defaultEndpoint}
 		}
 	}
 

db/keyval/etcd/plugin_impl_etcd.go

@@ -82,9 +82,11 @@ type Deps struct {
 // the connection cannot be established.
 func (p *Plugin) Init() (err error) {
 	// Read ETCD configuration file. Returns error if does not exists.
-	p.config, err = p.getEtcdConfig()
-	if err != nil || p.disabled {
-		return err
+	if p.config == nil {
+		p.config, err = p.getEtcdConfig()
+		if err != nil || p.disabled {
+			return err
+		}
 	}
 
 	// Transforms .yaml config to ETCD client configuration
@@ -284,16 +286,18 @@ func (p *Plugin) statusCheckProbe() (statuscheck.PluginState, error) {
 }
 
 func (p *Plugin) getEtcdConfig() (*Config, error) {
-	var etcdCfg Config
-	found, err := p.Cfg.LoadValue(&etcdCfg)
-	if err != nil {
-		return nil, err
-	}
-	if !found {
-		p.Log.Info("ETCD config not found, skip loading this plugin")
-		p.disabled = true
+	var etcdCfg = DefaultConfig()
+	if p.Cfg != nil {
+		found, err := p.Cfg.LoadValue(etcdCfg)
+		if err != nil {
+			return nil, err
+		}
+		if !found {
+			p.Log.Info("ETCD config not found, skip loading this plugin")
+			p.disabled = true
+		}
 	}
-	return &etcdCfg, nil
+	return etcdCfg, nil
 }
 
 func (p *Plugin) startPeriodicAutoCompact(period time.Duration) {