[VOL-2736]host and port should be specified as a single argument not as two separate arguments
Change-Id: I5a3a494c38dafa1e7e18e1f1cd55c0035359c7a9
diff --git a/rw_core/core/api/grpc_nbi_handler_test.go b/rw_core/core/api/grpc_nbi_handler_test.go
index 0fe877e..5801f27 100755
--- a/rw_core/core/api/grpc_nbi_handler_test.go
+++ b/rw_core/core/api/grpc_nbi_handler_test.go
@@ -24,6 +24,7 @@
"os"
"runtime"
"runtime/pprof"
+ "strconv"
"strings"
"sync"
"testing"
@@ -93,26 +94,23 @@
cfg.CoreTopic = "rw_core"
cfg.DefaultRequestTimeout = nb.defaultTimeout
cfg.DefaultCoreTimeout = nb.defaultTimeout
- cfg.KVStorePort = nb.kvClientPort
cfg.InCompetingMode = inCompeteMode
+ cfg.KVStoreAddress = "127.0.0.1" + ":" + strconv.Itoa(nb.kvClientPort)
grpcPort, err := freeport.GetFreePort()
if err != nil {
logger.Fatal("Cannot get a freeport for grpc")
}
- cfg.GrpcPort = grpcPort
- cfg.GrpcHost = "127.0.0.1"
+ cfg.GrpcAddress = "127.0.0.1" + ":" + strconv.Itoa(grpcPort)
setCoreCompeteMode(inCompeteMode)
client := tst.SetupKVClient(cfg, nb.coreInstanceID)
backend := &db.Backend{
Client: client,
StoreType: cfg.KVStoreType,
- Host: cfg.KVStoreHost,
- Port: cfg.KVStorePort,
+ Address: cfg.KVStoreAddress,
Timeout: cfg.KVStoreTimeout,
LivenessChannelInterval: cfg.LiveProbeInterval / 2}
nb.kmp = kafka.NewInterContainerProxy(
- kafka.InterContainerHost(cfg.KafkaAdapterHost),
- kafka.InterContainerPort(cfg.KafkaAdapterPort),
+ kafka.InterContainerAddress(cfg.KafkaAdapterAddress),
kafka.MsgClient(nb.kClient),
kafka.DefaultTopic(&kafka.Topic{Name: cfg.CoreTopic}),
kafka.DeviceDiscoveryTopic(&kafka.Topic{Name: cfg.AffinityRouterTopic}))
diff --git a/rw_core/core/core.go b/rw_core/core/core.go
index 0a76ca2..0cfa915 100644
--- a/rw_core/core/core.go
+++ b/rw_core/core/core.go
@@ -18,7 +18,6 @@
import (
"context"
- "strconv"
"time"
"github.com/opencord/voltha-go/db/model"
@@ -73,14 +72,14 @@
// setup kv client
logger.Debugw("create-kv-client", log.Fields{"kvstore": cf.KVStoreType})
- kvClient, err := newKVClient(cf.KVStoreType, cf.KVStoreHost+":"+strconv.Itoa(cf.KVStorePort), cf.KVStoreTimeout)
+ kvClient, err := newKVClient(cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout)
if err != nil {
logger.Fatal(err)
}
defer stopKVClient(context.Background(), kvClient)
// sync logging config with kv store
- cm := conf.NewConfigManager(kvClient, cf.KVStoreType, cf.KVStoreHost, cf.KVStorePort, cf.KVStoreTimeout)
+ cm := conf.NewConfigManager(kvClient, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout)
go conf.StartLogLevelConfigProcessing(cm, ctx)
backend := cm.Backend
@@ -94,8 +93,7 @@
// create kafka client
kafkaClient := kafka.NewSaramaClient(
- kafka.Host(cf.KafkaAdapterHost),
- kafka.Port(cf.KafkaAdapterPort),
+ kafka.Address(cf.KafkaAdapterAddress),
kafka.ConsumerType(kafka.GroupCustomer),
kafka.ProducerReturnOnErrors(true),
kafka.ProducerReturnOnSuccess(true),
@@ -119,7 +117,7 @@
// connect to kafka, then wait until reachable and publisher/consumer created
// core.kmp must be created before deviceMgr and adapterMgr
- kmp, err := startKafkInterContainerProxy(ctx, kafkaClient, cf.KafkaAdapterHost, cf.KafkaAdapterPort, cf.CoreTopic, cf.AffinityRouterTopic, cf.ConnectionRetryInterval)
+ kmp, err := startKafkInterContainerProxy(ctx, kafkaClient, cf.KafkaAdapterAddress, cf.CoreTopic, cf.AffinityRouterTopic, cf.ConnectionRetryInterval)
if err != nil {
logger.Warn("Failed to setup kafka connection")
return
@@ -135,7 +133,7 @@
registerAdapterRequestHandlers(kmp, deviceMgr, adapterMgr, cf.CoreTopic)
// start gRPC handler
- grpcServer := grpcserver.NewGrpcServer(cf.GrpcHost, cf.GrpcPort, nil, false, probe.GetProbeFromContext(ctx))
+ grpcServer := grpcserver.NewGrpcServer(cf.GrpcAddress, nil, false, probe.GetProbeFromContext(ctx))
go startGRPCService(ctx, grpcServer, api.NewNBIHandler(deviceMgr, logicalDeviceMgr, adapterMgr))
defer grpcServer.Stop()
diff --git a/rw_core/core/device/agent_test.go b/rw_core/core/device/agent_test.go
index cd49681..93277ff 100755
--- a/rw_core/core/device/agent_test.go
+++ b/rw_core/core/device/agent_test.go
@@ -20,6 +20,7 @@
"context"
"math/rand"
"sort"
+ "strconv"
"strings"
"sync"
"testing"
@@ -114,25 +115,22 @@
cfg := config.NewRWCoreFlags()
cfg.CoreTopic = "rw_core"
cfg.DefaultRequestTimeout = dat.defaultTimeout
- cfg.KVStorePort = dat.kvClientPort
+ cfg.KVStoreAddress = "127.0.0.1" + ":" + strconv.Itoa(dat.kvClientPort)
cfg.InCompetingMode = inCompeteMode
grpcPort, err := freeport.GetFreePort()
if err != nil {
logger.Fatal("Cannot get a freeport for grpc")
}
- cfg.GrpcPort = grpcPort
- cfg.GrpcHost = "127.0.0.1"
+ cfg.GrpcAddress = "127.0.0.1" + ":" + strconv.Itoa(grpcPort)
client := tst.SetupKVClient(cfg, dat.coreInstanceID)
backend := &db.Backend{
Client: client,
StoreType: cfg.KVStoreType,
- Host: cfg.KVStoreHost,
- Port: cfg.KVStorePort,
+ Address: cfg.KVStoreAddress,
Timeout: cfg.KVStoreTimeout,
LivenessChannelInterval: cfg.LiveProbeInterval / 2}
dat.kmp = kafka.NewInterContainerProxy(
- kafka.InterContainerHost(cfg.KafkaAdapterHost),
- kafka.InterContainerPort(cfg.KafkaAdapterPort),
+ kafka.InterContainerAddress(cfg.KafkaAdapterAddress),
kafka.MsgClient(dat.kClient),
kafka.DefaultTopic(&kafka.Topic{Name: cfg.CoreTopic}),
kafka.DeviceDiscoveryTopic(&kafka.Topic{Name: cfg.AffinityRouterTopic}))
diff --git a/rw_core/core/device/logical_agent_test.go b/rw_core/core/device/logical_agent_test.go
index 2649ee4..2e1b1d3 100644
--- a/rw_core/core/device/logical_agent_test.go
+++ b/rw_core/core/device/logical_agent_test.go
@@ -18,6 +18,7 @@
import (
"context"
"math/rand"
+ "strconv"
"sync"
"testing"
"time"
@@ -456,25 +457,22 @@
cfg := config.NewRWCoreFlags()
cfg.CoreTopic = "rw_core"
cfg.DefaultRequestTimeout = lda.defaultTimeout
- cfg.KVStorePort = lda.kvClientPort
+ cfg.KVStoreAddress = "127.0.0.1" + ":" + strconv.Itoa(lda.kvClientPort)
cfg.InCompetingMode = inCompeteMode
grpcPort, err := freeport.GetFreePort()
if err != nil {
logger.Fatal("Cannot get a freeport for grpc")
}
- cfg.GrpcPort = grpcPort
- cfg.GrpcHost = "127.0.0.1"
+ cfg.GrpcAddress = "127.0.0.1" + ":" + strconv.Itoa(grpcPort)
client := tst.SetupKVClient(cfg, lda.coreInstanceID)
backend := &db.Backend{
Client: client,
StoreType: cfg.KVStoreType,
- Host: cfg.KVStoreHost,
- Port: cfg.KVStorePort,
+ Address: cfg.KVStoreAddress,
Timeout: cfg.KVStoreTimeout,
LivenessChannelInterval: cfg.LiveProbeInterval / 2}
lda.kmp = kafka.NewInterContainerProxy(
- kafka.InterContainerHost(cfg.KafkaAdapterHost),
- kafka.InterContainerPort(cfg.KafkaAdapterPort),
+ kafka.InterContainerAddress(cfg.KafkaAdapterAddress),
kafka.MsgClient(lda.kClient),
kafka.DefaultTopic(&kafka.Topic{Name: cfg.CoreTopic}),
kafka.DeviceDiscoveryTopic(&kafka.Topic{Name: cfg.AffinityRouterTopic}))
diff --git a/rw_core/core/kafka.go b/rw_core/core/kafka.go
index 18b9ec8..3cb0292 100644
--- a/rw_core/core/kafka.go
+++ b/rw_core/core/kafka.go
@@ -29,15 +29,14 @@
)
// startKafkInterContainerProxy is responsible for starting the Kafka Interadapter Proxy
-func startKafkInterContainerProxy(ctx context.Context, kafkaClient kafka.Client, host string, port int, coreTopic, affinityRouterTopic string, connectionRetryInterval time.Duration) (kafka.InterContainerProxy, error) {
- logger.Infow("initialize-kafka-manager", log.Fields{"host": host, "port": port, "topic": coreTopic})
+func startKafkInterContainerProxy(ctx context.Context, kafkaClient kafka.Client, address string, coreTopic, affinityRouterTopic string, connectionRetryInterval time.Duration) (kafka.InterContainerProxy, error) {
+ logger.Infow("initialize-kafka-manager", log.Fields{"address": address, "topic": coreTopic})
probe.UpdateStatusFromContext(ctx, "message-bus", probe.ServiceStatusPreparing)
// create the kafka RPC proxy
kmp := kafka.NewInterContainerProxy(
- kafka.InterContainerHost(host),
- kafka.InterContainerPort(port),
+ kafka.InterContainerAddress(address),
kafka.MsgClient(kafkaClient),
kafka.DefaultTopic(&kafka.Topic{Name: coreTopic}),
kafka.DeviceDiscoveryTopic(&kafka.Topic{Name: affinityRouterTopic}))
@@ -45,8 +44,8 @@
probe.UpdateStatusFromContext(ctx, "message-bus", probe.ServiceStatusPrepared)
// wait for connectivity
- logger.Infow("starting-kafka-manager", log.Fields{"host": host,
- "port": port, "topic": coreTopic})
+ logger.Infow("starting-kafka-manager", log.Fields{"address": address,
+ "topic": coreTopic})
for {
// If we haven't started yet, then try to start