HDFSWriter microservice working copy

Issue-ID: ONAPARC-453
Signed-off-by: Rajamohan Raj <rajamohan.raj@intel.com>
Change-Id: I11c91b642e466763c1ca6f5734bf81fb260e2b39

5 files changed, 185 insertions, 0 deletions

diff --git a/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/hdfs-config.go b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/hdfs-config.go
new file mode 100644
index 00000000..ac33bc6a
--- /dev/null
+++ b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/hdfs-config.go
@@ -0,0 +1,37 @@
+package utils
+
+import (
+	"os"
+)
+
+// SetHdfsParametersByObjectMap set the value of the hdfs config parameters
+// and return HdfsConfig object
+func SetHdfsParametersByObjectMap(m map[string]interface{}) HdfsConfig{
+
+	hc := HdfsConfig{}
+	hc.hdfsURL = m["hdfs_url"].(string)
+	return hc
+
+}
+
+// SetHdfsParametersByEnvVariables sets the hdfs parameters
+func SetHdfsParametersByEnvVariables() HdfsConfig {
+	
+	slogger := GetLoggerInstance()
+	hdfsConfigObject := HdfsConfig{
+		hdfsURL: os.Getenv("HDFS_URL"),
+	}
+	slogger.Infof("::hdfsURL:: %s", hdfsConfigObject.hdfsURL)
+	return hdfsConfigObject
+	
+}
+
+// HdfsConfig contains hdfs related config items
+type HdfsConfig struct {
+	hdfsURL string
+}
+
+// GetHdfsURL returns HdfsURL
+func (h HdfsConfig) GetHdfsURL() string {
+	return h.hdfsURL
+}
\ No newline at end of file
diff --git a/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/hdfsUtils.go b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/hdfsUtils.go
new file mode 100644
index 00000000..1a93a5ad
--- /dev/null
+++ b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/hdfsUtils.go
@@ -0,0 +1,33 @@
+package utils
+
+import (
+	"fmt"
+	"github.com/colinmarc/hdfs"
+	//"sync"
+	//"go.uber.org/zap"
+)
+
+
+//var clientOnce sync.Once
+//var hdfsClient *hdfs.Client
+//var slogger *zap.SugaredLogger
+
+
+//GetHdfsClientInstance returns a singleton hdfsClient instance
+// func GetHdfsClientInstance(hdfsURL string) (*hdfs.Client){
+// 	clientOnce.Do(func(){
+// 		hdfsClient = createHdfsClient(hdfsURL)
+// 	})
+// 	return hdfsClient
+// }
+
+//CreateHdfsClient creates a hdfs client and returns hdfs client
+func CreateHdfsClient(hdfsURL string) (*hdfs.Client){
+	slogger := GetLoggerInstance()
+	hdfsClient, hdfsConnectError := hdfs.New(hdfsURL)
+	if hdfsConnectError !=nil {
+		slogger.Fatalf(":::Error in create hdfsClient::: %v", hdfsConnectError)
+		fmt.Printf("::Unable to initialize hdfsURL, check logs")
+	}
+	return hdfsClient
+}
\ No newline at end of file
diff --git a/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/kafka-config.go b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/kafka-config.go
new file mode 100644
index 00000000..080bfd4b
--- /dev/null
+++ b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/kafka-config.go
@@ -0,0 +1,55 @@
+package utils
+
+
+import (
+	"os"
+)
+
+// SetKafkaParametersByObjectMap sets the  value of the kafka parameters
+// and sets the KafkaConfig object 
+func SetKafkaParametersByObjectMap(m map[string]interface{}) KafkaConfig {
+	kc := KafkaConfig{}
+	kc.broker = m["broker"].(string)
+	kc.group = m["group"].(string)
+	kc.topic = m["topic"].(string)
+
+	return kc
+}
+
+// SetKafkaParametersByEnvVariables sets the kafka parameters
+func SetKafkaParametersByEnvVariables() KafkaConfig {
+	slogger := GetLoggerInstance()
+	
+	kafkaConfigObject := KafkaConfig{
+		broker: os.Getenv("BROKER"),
+		group: os.Getenv("GROUP"),
+		topic: os.Getenv("TOPIC"),
+	}
+	slogger.Infof("::broker:: %s", kafkaConfigObject.broker)
+	slogger.Infof("::group:: %s", kafkaConfigObject.group)
+	slogger.Infof("::topic:: %s", kafkaConfigObject.topic)
+
+	return kafkaConfigObject
+}
+
+// KafkaConfig contains all the config parameters needed for kafka. This can be extended over time
+type KafkaConfig struct {
+	broker string
+	group string
+	topic string
+}
+
+// GetBroker returns kafka broker configured
+func (k KafkaConfig) GetBroker() string {
+	return k.broker
+}
+
+// GetGroup returns kafka group configured
+func (k KafkaConfig) GetGroup() string {
+	return k.group
+}
+
+// GetTopic returns kafka topic configured
+func (k KafkaConfig) GetTopic() string {
+	return k.topic
+}
\ No newline at end of file
diff --git a/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/logutils.go b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/logutils.go
new file mode 100644
index 00000000..0f72e718
--- /dev/null
+++ b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/logutils.go
@@ -0,0 +1,32 @@
+package utils
+
+import (
+	"go.uber.org/zap"
+	"fmt"
+	"sync"
+)
+
+
+
+var logOnce sync.Once
+var logger *zap.SugaredLogger
+
+//GetLoggerInstance returns a singleton instance of logger
+func GetLoggerInstance() (*zap.SugaredLogger){
+	logOnce.Do(func(){
+		logger = createLogger()
+	})
+	return logger
+}
+
+
+//createLogger returns a SugaredLogger, sugaredLogger can be directly used to generate logs
+func createLogger() (*zap.SugaredLogger){
+	logger, err := zap.NewDevelopment()
+	if err != nil {
+		fmt.Printf("can't initialize zap logger: %v", err)
+	}
+	defer logger.Sync()
+	slogger := logger.Sugar()
+	return slogger
+}
\ No newline at end of file
diff --git a/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/readJson.go b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/readJson.go
new file mode 100644
index 00000000..bfab64e6
--- /dev/null
+++ b/vnfs/DAaaS/microservices/GoApps/src/go-hdfs-writer/pkg/utils/readJson.go
@@ -0,0 +1,28 @@
+package utils
+
+import (
+	"os"
+	"io/ioutil"
+)
+
+
+//ReadJSON reads the content of a give file and returns as a string
+// used for small config files only.
+func ReadJSON(path string) string {
+	slogger := GetLoggerInstance()
+	jsonFile, err := os.Open(path)
+	if err!=nil{
+		//fmt.Print(err)
+		slogger.Errorf("Unable to open file: %s", path)
+		slogger.Errorf("Error::: %s", err)
+
+	}else{
+		slogger.Infof("Successfully opened config.json")
+	}
+	
+	defer jsonFile.Close()
+	byteValue, _ := ioutil.ReadAll(jsonFile)
+	s := string(byteValue)
+	return s
+}
+