--- /dev/null
+// ============LICENSE_START===============================================
+// Copyright (C) 2023 Nordix Foundation. All rights reserved.
+// ========================================================================
+// 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.
+// ============LICENSE_END=================================================
+//
+
+package main
+
+import (
+ "bytes"
+ "compress/gzip"
+ "context"
+ "crypto/tls"
+ "encoding/json"
+ "encoding/xml"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "os/signal"
+ "reflect"
+ "strings"
+ "sync"
+ "syscall"
+
+ "net/http"
+ "os"
+ "runtime"
+ "strconv"
+ "time"
+
+ "github.com/google/uuid"
+ "golang.org/x/oauth2/clientcredentials"
+
+ log "github.com/sirupsen/logrus"
+
+ "github.com/gorilla/mux"
+
+ "net/http/pprof"
+
+ "github.com/confluentinc/confluent-kafka-go/kafka"
+ influxdb2 "github.com/influxdata/influxdb-client-go/v2"
+ jsoniter "github.com/json-iterator/go"
+ "github.com/minio/minio-go/v7"
+ "github.com/minio/minio-go/v7/pkg/credentials"
+)
+
+//== Constants ==//
+
+const http_port = 80
+const https_port = 443
+const config_file = "application_configuration.json"
+const server_crt = "server.crt"
+const server_key = "server.key"
+
+const producer_name = "kafka-producer"
+
+const registration_delay_short = 2
+const registration_delay_long = 120
+
+const mutexLocked = 1
+
+const (
+ Init AppStates = iota
+ Running
+ Terminating
+)
+
+const reader_queue_length = 100 //Per type job
+const writer_queue_length = 100 //Per info job
+const parallelism_limiter = 100 //For all jobs
+
+// This are optional - set if using SASL protocol is used towards kafka
+var creds_grant_type = os.Getenv("CREDS_GRANT_TYPE")
+var creds_client_secret = os.Getenv("CREDS_CLIENT_SECRET")
+var creds_client_id = os.Getenv("CREDS_CLIENT_ID")
+var creds_service_url = os.Getenv("AUTH_SERVICE_URL")
+
+//== Types ==//
+
+type AppStates int64
+
+type FilterParameters struct {
+ MeasuredEntityDns []string `json:"measuredEntityDns"`
+ MeasTypes []string `json:"measTypes"`
+ MeasObjClass []string `json:"measObjClass"`
+ MeasObjInstIds []string `json:"measObjInstIds"`
+}
+
+type InfoJobDataType struct {
+ InfoJobData struct {
+ KafkaOutputTopic string `json:"kafkaOutputTopic"`
+
+ DbUrl string `json:"db-url"`
+ DbOrg string `json:"db-org"`
+ DbBucket string `json:"db-bucket"`
+ DbToken string `json:"db-token"`
+
+ FilterParams FilterParameters `json:"filter"`
+ } `json:"info_job_data"`
+ InfoJobIdentity string `json:"info_job_identity"`
+ InfoTypeIdentity string `json:"info_type_identity"`
+ LastUpdated string `json:"last_updated"`
+ Owner string `json:"owner"`
+ TargetURI string `json:"target_uri"`
+}
+
+// Type for an infojob
+type InfoJobRecord struct {
+ job_info InfoJobDataType
+ output_topic string
+
+ statistics *InfoJobStats
+}
+
+// Type for an infojob
+type TypeJobRecord struct {
+ InfoType string
+ InputTopic string
+ data_in_channel chan *KafkaPayload
+ reader_control chan ReaderControl
+ job_control chan JobControl
+ groupId string
+ clientId string
+
+ statistics *TypeJobStats
+}
+
+// Type for controlling the topic reader
+type ReaderControl struct {
+ command string
+}
+
+// Type for controlling the topic writer
+type WriterControl struct {
+ command string
+}
+
+// Type for controlling the job
+type JobControl struct {
+ command string
+ filter Filter
+}
+
+type KafkaPayload struct {
+ msg *kafka.Message
+ topic string
+ jobid string
+}
+
+type FilterMaps struct {
+ sourceNameMap map[string]bool
+ measObjClassMap map[string]bool
+ measObjInstIdsMap map[string]bool
+ measTypesMap map[string]bool
+}
+
+type InfluxJobParameters struct {
+ DbUrl string
+ DbOrg string
+ DbBucket string
+ DbToken string
+}
+
+type Filter struct {
+ JobId string
+ OutputTopic string
+ filter FilterMaps
+
+ influxParameters InfluxJobParameters
+}
+
+// Type for info job statistics
+type InfoJobStats struct {
+ out_msg_cnt int
+ out_data_vol int64
+}
+
+// Type for type job statistics
+type TypeJobStats struct {
+ in_msg_cnt int
+ in_data_vol int64
+}
+
+// == API Datatypes ==//
+// Type for supported data types
+type DataType struct {
+ ID string `json:"id"`
+ KafkaInputTopic string `json:"kafkaInputTopic"`
+ InputJobType string `json:inputJobType`
+ InputJobDefinition struct {
+ KafkaOutputTopic string `json:kafkaOutputTopic`
+ } `json:inputJobDefinition`
+
+ ext_job *[]byte
+ ext_job_created bool
+ ext_job_id string
+}
+
+type DataTypes struct {
+ ProdDataTypes []DataType `json:"types"`
+}
+
+type Minio_buckets struct {
+ Buckets map[string]bool
+}
+
+//== External data types ==//
+
+// // Data type for event xml file download
+type XmlFileEventHeader struct {
+ ProductName string `json:"productName"`
+ VendorName string `json:"vendorName"`
+ Location string `json:"location"`
+ Compression string `json:"compression"`
+ SourceName string `json:"sourceName"`
+ FileFormatType string `json:"fileFormatType"`
+ FileFormatVersion string `json:"fileFormatVersion"`
+ StartEpochMicrosec int64 `json:"startEpochMicrosec"`
+ LastEpochMicrosec int64 `json:"lastEpochMicrosec"`
+ Name string `json:"name"`
+ ChangeIdentifier string `json:"changeIdentifier"`
+ InternalLocation string `json:"internalLocation"`
+ TimeZoneOffset string `json:"timeZoneOffset"`
+ //ObjectStoreBucket string `json:"objectStoreBucket"`
+}
+
+// Data types for input xml file
+type MeasCollecFile struct {
+ XMLName xml.Name `xml:"measCollecFile"`
+ Text string `xml:",chardata"`
+ Xmlns string `xml:"xmlns,attr"`
+ Xsi string `xml:"xsi,attr"`
+ SchemaLocation string `xml:"schemaLocation,attr"`
+ FileHeader struct {
+ Text string `xml:",chardata"`
+ FileFormatVersion string `xml:"fileFormatVersion,attr"`
+ VendorName string `xml:"vendorName,attr"`
+ DnPrefix string `xml:"dnPrefix,attr"`
+ FileSender struct {
+ Text string `xml:",chardata"`
+ LocalDn string `xml:"localDn,attr"`
+ ElementType string `xml:"elementType,attr"`
+ } `xml:"fileSender"`
+ MeasCollec struct {
+ Text string `xml:",chardata"`
+ BeginTime string `xml:"beginTime,attr"`
+ } `xml:"measCollec"`
+ } `xml:"fileHeader"`
+ MeasData struct {
+ Text string `xml:",chardata"`
+ ManagedElement struct {
+ Text string `xml:",chardata"`
+ LocalDn string `xml:"localDn,attr"`
+ SwVersion string `xml:"swVersion,attr"`
+ } `xml:"managedElement"`
+ MeasInfo []struct {
+ Text string `xml:",chardata"`
+ MeasInfoId string `xml:"measInfoId,attr"`
+ Job struct {
+ Text string `xml:",chardata"`
+ JobId string `xml:"jobId,attr"`
+ } `xml:"job"`
+ GranPeriod struct {
+ Text string `xml:",chardata"`
+ Duration string `xml:"duration,attr"`
+ EndTime string `xml:"endTime,attr"`
+ } `xml:"granPeriod"`
+ RepPeriod struct {
+ Text string `xml:",chardata"`
+ Duration string `xml:"duration,attr"`
+ } `xml:"repPeriod"`
+ MeasType []struct {
+ Text string `xml:",chardata"`
+ P string `xml:"p,attr"`
+ } `xml:"measType"`
+ MeasValue []struct {
+ Text string `xml:",chardata"`
+ MeasObjLdn string `xml:"measObjLdn,attr"`
+ R []struct {
+ Text string `xml:",chardata"`
+ P string `xml:"p,attr"`
+ } `xml:"r"`
+ Suspect string `xml:"suspect"`
+ } `xml:"measValue"`
+ } `xml:"measInfo"`
+ } `xml:"measData"`
+ FileFooter struct {
+ Text string `xml:",chardata"`
+ MeasCollec struct {
+ Text string `xml:",chardata"`
+ EndTime string `xml:"endTime,attr"`
+ } `xml:"measCollec"`
+ } `xml:"fileFooter"`
+}
+
+// Data type for json file
+// Splitted in sevreal part to allow add/remove in lists
+type MeasResults struct {
+ P int `json:"p"`
+ SValue string `json:"sValue"`
+}
+
+type MeasValues struct {
+ MeasObjInstID string `json:"measObjInstId"`
+ SuspectFlag string `json:"suspectFlag"`
+ MeasResultsList []MeasResults `json:"measResults"`
+}
+
+type SMeasTypes struct {
+ SMeasType string `json:"sMeasTypesList"`
+}
+
+type MeasInfoList struct {
+ MeasInfoID struct {
+ SMeasInfoID string `json:"sMeasInfoId"`
+ } `json:"measInfoId"`
+ MeasTypes struct {
+ SMeasTypesList []string `json:"sMeasTypesList"`
+ } `json:"measTypes"`
+ MeasValuesList []MeasValues `json:"measValuesList"`
+}
+
+type PMJsonFile struct {
+ Event struct {
+ CommonEventHeader struct {
+ Domain string `json:"domain"`
+ EventID string `json:"eventId"`
+ Sequence int `json:"sequence"`
+ EventName string `json:"eventName"`
+ SourceName string `json:"sourceName"`
+ ReportingEntityName string `json:"reportingEntityName"`
+ Priority string `json:"priority"`
+ StartEpochMicrosec int64 `json:"startEpochMicrosec"`
+ LastEpochMicrosec int64 `json:"lastEpochMicrosec"`
+ Version string `json:"version"`
+ VesEventListenerVersion string `json:"vesEventListenerVersion"`
+ TimeZoneOffset string `json:"timeZoneOffset"`
+ } `json:"commonEventHeader"`
+ Perf3GppFields struct {
+ Perf3GppFieldsVersion string `json:"perf3gppFieldsVersion"`
+ MeasDataCollection struct {
+ GranularityPeriod int `json:"granularityPeriod"`
+ MeasuredEntityUserName string `json:"measuredEntityUserName"`
+ MeasuredEntityDn string `json:"measuredEntityDn"`
+ MeasuredEntitySoftwareVersion string `json:"measuredEntitySoftwareVersion"`
+ SMeasInfoList []MeasInfoList `json:"measInfoList"`
+ } `json:"measDataCollection"`
+ } `json:"perf3gppFields"`
+ } `json:"event"`
+}
+
+// Data type for converted json file message
+type FileDownloadedEvt struct {
+ Filename string `json:"filename"`
+}
+
+//== Variables ==//
+
+var AppState = Init
+
+// Lock for all internal data
+var datalock sync.Mutex
+
+var producer_instance_name string = producer_name
+
+// Keep all info type jobs, key == type id
+var TypeJobs map[string]TypeJobRecord = make(map[string]TypeJobRecord)
+
+// Keep all info jobs, key == job id
+var InfoJobs map[string]InfoJobRecord = make(map[string]InfoJobRecord)
+
+var InfoTypes DataTypes
+
+// Limiter - valid for all jobs
+var jobLimiterChan = make(chan struct{}, parallelism_limiter)
+
+// TODO: Config param?
+var bucket_location = "swe"
+
+var httpclient = &http.Client{}
+
+// == Env variables ==//
+var bootstrapserver = os.Getenv("KAFKA_SERVER")
+var files_volume = os.Getenv("FILES_VOLUME")
+var ics_server = os.Getenv("ICS")
+var self = os.Getenv("SELF")
+var filestore_user = os.Getenv("FILESTORE_USER")
+var filestore_pwd = os.Getenv("FILESTORE_PWD")
+var filestore_server = os.Getenv("FILESTORE_SERVER")
+
+var data_out_channel = make(chan *KafkaPayload, writer_queue_length)
+var writer_control = make(chan WriterControl, 1)
+
+var minio_bucketlist map[string]Minio_buckets = make(map[string]Minio_buckets)
+
+// == Main ==//
+func main() {
+
+ //log.SetLevel(log.InfoLevel)
+ log.SetLevel(log.TraceLevel)
+
+ log.Info("Server starting...")
+
+ if self == "" {
+ log.Panic("Env SELF not configured")
+ }
+ if bootstrapserver == "" {
+ log.Panic("Env KAFKA_SERVER not set")
+ }
+ if ics_server == "" {
+ log.Panic("Env ICS not set")
+ }
+ if os.Getenv("KP") != "" {
+ producer_instance_name = producer_instance_name + "-" + os.Getenv("KP")
+ }
+
+ rtr := mux.NewRouter()
+ rtr.HandleFunc("/callbacks/job/"+producer_instance_name, create_job)
+ rtr.HandleFunc("/callbacks/job/"+producer_instance_name+"/{job_id}", delete_job)
+ rtr.HandleFunc("/callbacks/supervision/"+producer_instance_name, supervise_producer)
+ rtr.HandleFunc("/statistics", statistics)
+ rtr.HandleFunc("/logging/{level}", logging_level)
+ rtr.HandleFunc("/logging", logging_level)
+ rtr.HandleFunc("/", alive)
+
+ //For perf/mem profiling
+ rtr.HandleFunc("/custom_debug_path/profile", pprof.Profile)
+
+ http.Handle("/", rtr)
+
+ http_server := &http.Server{Addr: ":" + strconv.Itoa(http_port), Handler: nil}
+
+ cer, err := tls.LoadX509KeyPair(server_crt, server_key)
+ if err != nil {
+ log.Error("Cannot load key and cert - %v\n", err)
+ return
+ }
+ config := &tls.Config{Certificates: []tls.Certificate{cer}}
+ https_server := &http.Server{Addr: ":" + strconv.Itoa(https_port), TLSConfig: config, Handler: nil}
+
+ //TODO: Make http on/off configurable
+ // Run http
+ go func() {
+ log.Info("Starting http service...")
+ err := http_server.ListenAndServe()
+ if err == http.ErrServerClosed { // graceful shutdown
+ log.Info("http server shutdown...")
+ } else if err != nil {
+ log.Error("http server error: %v\n", err)
+ }
+ }()
+
+ //TODO: Make https on/off configurable
+ // Run https
+ go func() {
+ log.Info("Starting https service...")
+ err := https_server.ListenAndServe()
+ if err == http.ErrServerClosed { // graceful shutdown
+ log.Info("https server shutdown...")
+ } else if err != nil {
+ log.Error("https server error: %v\n", err)
+ }
+ }()
+ check_tcp(strconv.Itoa(http_port))
+ check_tcp(strconv.Itoa(https_port))
+
+ go start_topic_writer(writer_control, data_out_channel)
+
+ //Setup proc for periodic type registration
+ var event_chan = make(chan int) //Channel for stopping the proc
+ go periodic_registration(event_chan)
+
+ //Wait for term/int signal do try to shut down gracefully
+ sigs := make(chan os.Signal, 1)
+ signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+ go func() {
+ sig := <-sigs
+ fmt.Printf("Received signal %s - application will terminate\n", sig)
+ event_chan <- 0 // Stop periodic registration
+ datalock.Lock()
+ defer datalock.Unlock()
+ AppState = Terminating
+ http_server.Shutdown(context.Background())
+ https_server.Shutdown(context.Background())
+ // Stopping jobs
+ for key, _ := range TypeJobs {
+ log.Info("Stopping type job:", key)
+ for _, dp := range InfoTypes.ProdDataTypes {
+ if key == dp.ID {
+ remove_type_job(dp)
+ }
+ }
+ }
+ }()
+
+ AppState = Running
+
+ //Wait until all go routines has exited
+ runtime.Goexit()
+
+ fmt.Println("main routine exit")
+ fmt.Println("server stopped")
+}
+
+func check_tcp(port string) {
+ log.Info("Checking tcp port: ", port)
+ for true {
+ address := net.JoinHostPort("localhost", port)
+ // 3 second timeout
+ conn, err := net.DialTimeout("tcp", address, 3*time.Second)
+ if err != nil {
+ log.Info("Checking tcp port: ", port, " failed, retrying...")
+ } else {
+ if conn != nil {
+ log.Info("Checking tcp port: ", port, " - OK")
+ _ = conn.Close()
+ return
+ } else {
+ log.Info("Checking tcp port: ", port, " failed, retrying...")
+ }
+ }
+ }
+}
+
+//== Core functions ==//
+
+// Run periodic registration of producers
+func periodic_registration(evtch chan int) {
+ var delay int = 1
+ for {
+ select {
+ case msg := <-evtch:
+ if msg == 0 { // Stop thread
+ return
+ }
+ case <-time.After(time.Duration(delay) * time.Second):
+ ok := register_producer()
+ if ok {
+ delay = registration_delay_long
+ } else {
+ if delay < registration_delay_long {
+ delay += registration_delay_short
+ } else {
+ delay = registration_delay_short
+ }
+ }
+ }
+ }
+}
+
+func register_producer() bool {
+
+ log.Info("Registering producer: ", producer_instance_name)
+
+ file, err := os.ReadFile(config_file)
+ if err != nil {
+ log.Error("Cannot read config file: ", config_file)
+ log.Error("Registering producer: ", producer_instance_name, " - failed")
+ return false
+ }
+ data := DataTypes{}
+ err = jsoniter.Unmarshal([]byte(file), &data)
+ if err != nil {
+ log.Error("Cannot parse config file: ", config_file)
+ log.Error("Registering producer: ", producer_instance_name, " - failed")
+ return false
+ }
+ var new_type_names []string
+
+ for i := 0; i < len(data.ProdDataTypes); i++ {
+ t1 := make(map[string]interface{})
+ t2 := make(map[string]interface{})
+
+ t2["schema"] = "http://json-schema.org/draft-07/schema#"
+ t2["title"] = data.ProdDataTypes[i].ID
+ t2["description"] = data.ProdDataTypes[i].ID
+ t2["type"] = "object"
+
+ t1["info_job_data_schema"] = t2
+
+ json, err := json.Marshal(t1)
+ if err != nil {
+ log.Error("Cannot create json for type: ", data.ProdDataTypes[i].ID)
+ log.Error("Registering producer: ", producer_instance_name, " - failed")
+ return false
+ } else {
+ //TODO: http/https should be configurable
+ ok := send_http_request(json, http.MethodPut, "http://"+ics_server+"/data-producer/v1/info-types/"+data.ProdDataTypes[i].ID, true, creds_grant_type != "")
+ if !ok {
+ log.Error("Cannot register type: ", data.ProdDataTypes[i].ID)
+ log.Error("Registering producer: ", producer_instance_name, " - failed")
+ return false
+ }
+ new_type_names = append(new_type_names, data.ProdDataTypes[i].ID)
+ }
+
+ }
+
+ log.Debug("Registering types: ", new_type_names)
+ m := make(map[string]interface{})
+ m["supported_info_types"] = new_type_names
+ //TODO: http/https should be configurable
+ m["info_job_callback_url"] = "http://" + self + "/callbacks/job/" + producer_instance_name
+ //TODO: http/https should be configurable
+ m["info_producer_supervision_callback_url"] = "http://" + self + "/callbacks/supervision/" + producer_instance_name
+
+ json, err := json.Marshal(m)
+ if err != nil {
+ log.Error("Cannot create json for producer: ", producer_instance_name)
+ log.Error("Registering producer: ", producer_instance_name, " - failed")
+ return false
+ }
+ ok := send_http_request(json, http.MethodPut, "http://"+ics_server+"/data-producer/v1/info-producers/"+producer_instance_name, true, creds_grant_type != "")
+ if !ok {
+ log.Error("Cannot register producer: ", producer_instance_name)
+ log.Error("Registering producer: ", producer_instance_name, " - failed")
+ return false
+ }
+ datalock.Lock()
+ defer datalock.Unlock()
+
+ var current_type_names []string
+ for _, v := range InfoTypes.ProdDataTypes {
+ current_type_names = append(current_type_names, v.ID)
+ if contains_str(new_type_names, v.ID) {
+ //Type exist
+ log.Debug("Type ", v.ID, " exists")
+ create_ext_job(v)
+ } else {
+ //Type is removed
+ log.Info("Removing type job for type: ", v.ID, " Type not in configuration")
+ remove_type_job(v)
+ }
+ }
+
+ for _, v := range data.ProdDataTypes {
+ if contains_str(current_type_names, v.ID) {
+ //Type exist
+ log.Debug("Type ", v.ID, " exists")
+ create_ext_job(v)
+ } else {
+ //Type is new
+ log.Info("Adding type job for type: ", v.ID, " Type added to configuration")
+ start_type_job(v)
+ }
+ }
+
+ InfoTypes = data
+ log.Debug("Datatypes: ", InfoTypes)
+
+ log.Info("Registering producer: ", producer_instance_name, " - OK")
+ return true
+}
+
+func remove_type_job(dp DataType) {
+ log.Info("Removing type job: ", dp.ID)
+ j, ok := TypeJobs[dp.ID]
+ if ok {
+ j.reader_control <- ReaderControl{"EXIT"}
+ }
+
+ if dp.ext_job_created == true {
+ dp.ext_job_id = dp.InputJobType + "_" + generate_uuid_from_type(dp.InputJobType)
+ //TODO: http/https should be configurable
+ ok := send_http_request(*dp.ext_job, http.MethodDelete, "http://"+ics_server+"/data-consumer/v1/info-jobs/"+dp.ext_job_id, true, creds_grant_type != "")
+ if !ok {
+ log.Error("Cannot delete job: ", dp.ext_job_id)
+ }
+ dp.ext_job_created = false
+ dp.ext_job = nil
+ }
+
+}
+
+func start_type_job(dp DataType) {
+ log.Info("Starting type job: ", dp.ID)
+ job_record := TypeJobRecord{}
+
+ job_record.job_control = make(chan JobControl, 1)
+ job_record.reader_control = make(chan ReaderControl, 1)
+ job_record.data_in_channel = make(chan *KafkaPayload, reader_queue_length)
+ job_record.InfoType = dp.ID
+ job_record.InputTopic = dp.KafkaInputTopic
+ job_record.groupId = "kafka-procon-" + dp.ID
+ job_record.clientId = dp.ID + "-" + os.Getenv("KP")
+ var stats TypeJobStats
+ job_record.statistics = &stats
+
+ switch dp.ID {
+ case "xml-file-data-to-filestore":
+ go start_job_xml_file_data(dp.ID, job_record.job_control, job_record.data_in_channel, data_out_channel, "", "pm-files-json")
+ case "xml-file-data":
+ go start_job_xml_file_data(dp.ID, job_record.job_control, job_record.data_in_channel, data_out_channel, files_volume, "")
+ case "json-file-data-from-filestore":
+ go start_job_json_file_data(dp.ID, job_record.job_control, job_record.data_in_channel, data_out_channel, true)
+ case "json-file-data":
+ go start_job_json_file_data(dp.ID, job_record.job_control, job_record.data_in_channel, data_out_channel, false)
+ case "json-file-data-from-filestore-to-influx":
+ go start_job_json_file_data_influx(dp.ID, job_record.job_control, job_record.data_in_channel, true)
+ // case "json-data-to-influx":
+ // go start_job_json_data_influx(dp.ID, job_record.job_control, job_record.data_in_channel)
+
+ default:
+ }
+
+ go start_topic_reader(dp.KafkaInputTopic, dp.ID, job_record.reader_control, job_record.data_in_channel, job_record.groupId, job_record.clientId, &stats)
+
+ TypeJobs[dp.ID] = job_record
+ log.Debug("Type job input type: ", dp.InputJobType)
+ create_ext_job(dp)
+}
+
+func create_ext_job(dp DataType) {
+ if dp.InputJobType != "" {
+ jb := make(map[string]interface{})
+ jb["info_type_id"] = dp.InputJobType
+ jb["job_owner"] = "console" //TODO:
+ jb["status_notification_uri"] = "http://callback:80/post"
+ jb1 := make(map[string]interface{})
+ jb["job_definition"] = jb1
+ jb1["kafkaOutputTopic"] = dp.InputJobDefinition.KafkaOutputTopic
+
+ json, err := json.Marshal(jb)
+ dp.ext_job_created = false
+ dp.ext_job = nil
+ if err != nil {
+ log.Error("Cannot create json for type: ", dp.InputJobType)
+ return
+ }
+
+ dp.ext_job_id = dp.InputJobType + "_" + generate_uuid_from_type(dp.InputJobType)
+ //TODO: http/https should be configurable
+ ok := false
+ for !ok {
+ ok = send_http_request(json, http.MethodPut, "http://"+ics_server+"/data-consumer/v1/info-jobs/"+dp.ext_job_id, true, creds_grant_type != "")
+ if !ok {
+ log.Error("Cannot register job: ", dp.InputJobType)
+ //TODO: Restart after long time?
+ }
+ }
+ log.Debug("Registered job ok: ", dp.InputJobType)
+ dp.ext_job_created = true
+ dp.ext_job = &json
+ }
+}
+
+func remove_info_job(jobid string) {
+ log.Info("Removing info job: ", jobid)
+ filter := Filter{}
+ filter.JobId = jobid
+ jc := JobControl{}
+ jc.command = "REMOVE-FILTER"
+ jc.filter = filter
+ infoJob := InfoJobs[jobid]
+ typeJob := TypeJobs[infoJob.job_info.InfoTypeIdentity]
+ typeJob.job_control <- jc
+ delete(InfoJobs, jobid)
+
+}
+
+// == Helper functions ==//
+
+// Function to check the status of a mutex lock
+func MutexLocked(m *sync.Mutex) bool {
+ state := reflect.ValueOf(m).Elem().FieldByName("state")
+ return state.Int()&mutexLocked == mutexLocked
+}
+
+// Test if slice contains a string
+func contains_str(s []string, e string) bool {
+ for _, a := range s {
+ if a == e {
+ return true
+ }
+ }
+ return false
+}
+
+// Send a http request with json (json may be nil)
+func send_http_request(json []byte, method string, url string, retry bool, useAuth bool) bool {
+
+ // set the HTTP method, url, and request body
+ var req *http.Request
+ var err error
+ if json == nil {
+ req, err = http.NewRequest(method, url, http.NoBody)
+ } else {
+ req, err = http.NewRequest(method, url, bytes.NewBuffer(json))
+ req.Header.Set("Content-Type", "application/json; charset=utf-8")
+ }
+ if err != nil {
+ log.Error("Cannot create http request, method: ", method, " url: ", url)
+ return false
+ }
+
+ if useAuth {
+ token, err := fetch_token()
+ if err != nil {
+ log.Error("Cannot fetch token for http request: ", err)
+ return false
+ }
+ req.Header.Set("Authorization", "Bearer "+token.TokenValue)
+ }
+
+ log.Debug("HTTP request: ", req)
+
+ log.Debug("Sending http request")
+ resp, err2 := httpclient.Do(req)
+ if err2 != nil {
+ log.Error("Http request error: ", err2)
+ log.Error("Cannot send http request method: ", method, " url: ", url)
+ } else {
+ if resp.StatusCode == 200 || resp.StatusCode == 201 || resp.StatusCode == 204 {
+ log.Debug("Accepted http status: ", resp.StatusCode)
+ resp.Body.Close()
+ return true
+ }
+ log.Debug("HTTP resp: ", resp)
+ resp.Body.Close()
+ }
+ return false
+}
+
+// // Send a http request with json (json may be nil)
+// func send_http_request(json []byte, method string, url string, retry bool, useAuth bool) bool {
+
+// // set the HTTP method, url, and request body
+// var req *http.Request
+// var err error
+// if json == nil {
+// req, err = http.NewRequest(method, url, http.NoBody)
+// } else {
+// req, err = http.NewRequest(method, url, bytes.NewBuffer(json))
+// req.Header.Set("Content-Type", "application/json; charset=utf-8")
+// }
+// if err != nil {
+// log.Error("Cannot create http request, method: ", method, " url: ", url)
+// return false
+// }
+
+// if useAuth {
+// token, err := fetch_token()
+// if err != nil {
+// log.Error("Cannot fetch token for http request: ", err)
+// return false
+// }
+// req.Header.Set("Authorization", "Bearer "+token.TokenValue)
+// }
+
+// log.Debug("HTTP request: ", req)
+
+// retries := 1
+// if retry {
+// retries = 5
+// }
+// sleep_time := 1
+// for i := retries; i > 0; i-- {
+// log.Debug("Sending http request")
+// resp, err2 := httpclient.Do(req)
+// if err2 != nil {
+// log.Error("Http request error: ", err2)
+// log.Error("Cannot send http request method: ", method, " url: ", url, " - retries left: ", i-1)
+
+// time.Sleep(time.Duration(sleep_time) * time.Second)
+// sleep_time = 2 * sleep_time
+// } else {
+// if resp.StatusCode == 200 || resp.StatusCode == 201 || resp.StatusCode == 204 {
+// log.Debug("Accepted http status: ", resp.StatusCode)
+// resp.Body.Close()
+// return true
+// }
+// log.Debug("HTTP resp: ", resp)
+// resp.Body.Close()
+// }
+// }
+// return false
+// }
+
+// // Send a http request with json (json may be nil)
+// func send_http_request(json []byte, method string, url string, retry bool, useAuth bool) bool {
+// // initialize http client
+// client := &http.Client{}
+
+// // set the HTTP method, url, and request body
+// var req *http.Request
+// var err error
+// if json == nil {
+// req, err = http.NewRequest(method, url, http.NoBody)
+// } else {
+// req, err = http.NewRequest(method, url, bytes.NewBuffer(json))
+// req.Header.Set("Content-Type", "application/json; charset=utf-8")
+// }
+// if err != nil {
+// log.Error("Cannot create http request method: ", method, " url: ", url)
+// return false
+// }
+
+// useAuth = false
+// if useAuth {
+// token, err := fetch_token()
+// if err != nil {
+// log.Error("Cannot fetch token for http request: ", err)
+// return false
+// }
+// req.Header.Add("Authorization", "Bearer "+token.TokenValue)
+// }
+// log.Debug("HTTP request: ", req)
+
+// b, berr := io.ReadAll(req.Body)
+// if berr == nil {
+// log.Debug("HTTP request body length: ", len(b))
+// } else {
+// log.Debug("HTTP request - cannot check body length: ", berr)
+// }
+// if json == nil {
+// log.Debug("HTTP request null json")
+// } else {
+// log.Debug("HTTP request json: ", string(json))
+// }
+// requestDump, cerr := httputil.DumpRequestOut(req, true)
+// if cerr != nil {
+// fmt.Println(cerr)
+// }
+// fmt.Println(string(requestDump))
+
+// retries := 1
+// if retry {
+// retries = 5
+// }
+// sleep_time := 1
+// for i := retries; i > 0; i-- {
+// resp, err2 := client.Do(req)
+// if err2 != nil {
+// log.Error("Http request error: ", err2)
+// log.Error("Cannot send http request method: ", method, " url: ", url, " - retries left: ", i)
+
+// time.Sleep(time.Duration(sleep_time) * time.Second)
+// sleep_time = 2 * sleep_time
+// } else {
+// if resp.StatusCode == 200 || resp.StatusCode == 201 || resp.StatusCode == 204 {
+// log.Debug("Accepted http status: ", resp.StatusCode)
+// defer resp.Body.Close()
+// return true
+// }
+// }
+// }
+// return false
+// }
+
+func fetch_token() (*kafka.OAuthBearerToken, error) {
+ log.Debug("Get token inline")
+ conf := &clientcredentials.Config{
+ ClientID: creds_client_id,
+ ClientSecret: creds_client_secret,
+ TokenURL: creds_service_url,
+ }
+ token, err := conf.Token(context.Background())
+ if err != nil {
+ log.Warning("Cannot fetch access token: ", err)
+ return nil, err
+ }
+ extensions := map[string]string{}
+ log.Debug("=====================================================")
+ log.Debug("token: ", token)
+ log.Debug("=====================================================")
+ log.Debug("TokenValue: ", token.AccessToken)
+ log.Debug("=====================================================")
+ log.Debug("Expiration: ", token.Expiry)
+ t := token.Expiry
+ // t := token.Expiry.Add(-time.Minute)
+ // log.Debug("Modified expiration: ", t)
+ oauthBearerToken := kafka.OAuthBearerToken{
+ TokenValue: token.AccessToken,
+ Expiration: t,
+ Extensions: extensions,
+ }
+
+ return &oauthBearerToken, nil
+}
+
+// Function to print memory details
+// https://pkg.go.dev/runtime#MemStats
+func PrintMemUsage() {
+ if log.IsLevelEnabled(log.DebugLevel) || log.IsLevelEnabled(log.TraceLevel) {
+ var m runtime.MemStats
+ runtime.ReadMemStats(&m)
+ fmt.Printf("Alloc = %v MiB", bToMb(m.Alloc))
+ fmt.Printf("\tTotalAlloc = %v MiB", bToMb(m.TotalAlloc))
+ fmt.Printf("\tSys = %v MiB", bToMb(m.Sys))
+ fmt.Printf("\tNumGC = %v\n", m.NumGC)
+ fmt.Printf("HeapSys = %v MiB", bToMb(m.HeapSys))
+ fmt.Printf("\tStackSys = %v MiB", bToMb(m.StackSys))
+ fmt.Printf("\tMSpanSys = %v MiB", bToMb(m.MSpanSys))
+ fmt.Printf("\tMCacheSys = %v MiB", bToMb(m.MCacheSys))
+ fmt.Printf("\tBuckHashSys = %v MiB", bToMb(m.BuckHashSys))
+ fmt.Printf("\tGCSys = %v MiB", bToMb(m.GCSys))
+ fmt.Printf("\tOtherSys = %v MiB\n", bToMb(m.OtherSys))
+ }
+}
+
+func bToMb(b uint64) uint64 {
+ return b / 1024 / 1024
+}
+
+func generate_uuid_from_type(s string) string {
+ if len(s) > 16 {
+ s = s[:16]
+ }
+ for len(s) < 16 {
+ s = s + "0"
+ }
+ b := []byte(s)
+ b = b[:16]
+ uuid, _ := uuid.FromBytes(b)
+ return uuid.String()
+}
+
+// Write gzipped data to a Writer
+func gzipWrite(w io.Writer, data *[]byte) error {
+ gw, err1 := gzip.NewWriterLevel(w, gzip.BestSpeed)
+
+ if err1 != nil {
+ return err1
+ }
+ defer gw.Close()
+ _, err2 := gw.Write(*data)
+ return err2
+}
+
+// Write gunzipped data from Reader to a Writer
+func gunzipReaderToWriter(w io.Writer, data io.Reader) error {
+ gr, err1 := gzip.NewReader(data)
+
+ if err1 != nil {
+ return err1
+ }
+ defer gr.Close()
+ data2, err2 := io.ReadAll(gr)
+ if err2 != nil {
+ return err2
+ }
+ _, err3 := w.Write(data2)
+ if err3 != nil {
+ return err3
+ }
+ return nil
+}
+
+func create_minio_bucket(mc *minio.Client, client_id string, bucket string) error {
+ tctx := context.Background()
+ err := mc.MakeBucket(tctx, bucket, minio.MakeBucketOptions{Region: bucket_location})
+ if err != nil {
+ // Check to see if we already own this bucket (which happens if you run this twice)
+ exists, errBucketExists := mc.BucketExists(tctx, bucket)
+ if errBucketExists == nil && exists {
+ log.Debug("Already own bucket:", bucket)
+ add_bucket(client_id, bucket)
+ return nil
+ } else {
+ log.Error("Cannot create or check bucket ", bucket, " in minio client", err)
+ return err
+ }
+ }
+ log.Debug("Successfully created bucket: ", bucket)
+ add_bucket(client_id, bucket)
+ return nil
+}
+
+func check_minio_bucket(mc *minio.Client, client_id string, bucket string) bool {
+ ok := bucket_exist(client_id, bucket)
+ if ok {
+ return true
+ }
+ tctx := context.Background()
+ exists, err := mc.BucketExists(tctx, bucket)
+ if err == nil && exists {
+ log.Debug("Already own bucket:", bucket)
+ return true
+ }
+ log.Error("Bucket does not exist, bucket ", bucket, " in minio client", err)
+ return false
+}
+
+func add_bucket(minio_id string, bucket string) {
+ datalock.Lock()
+ defer datalock.Unlock()
+
+ b, ok := minio_bucketlist[minio_id]
+ if !ok {
+ b = Minio_buckets{}
+ b.Buckets = make(map[string]bool)
+ }
+ b.Buckets[bucket] = true
+ minio_bucketlist[minio_id] = b
+}
+
+func bucket_exist(minio_id string, bucket string) bool {
+ datalock.Lock()
+ defer datalock.Unlock()
+
+ b, ok := minio_bucketlist[minio_id]
+ if !ok {
+ return false
+ }
+ _, ok = b.Buckets[bucket]
+ return ok
+}
+
+//== http api functions ==//
+
+// create/update job
+func create_job(w http.ResponseWriter, req *http.Request) {
+ log.Debug("Create job, http method: ", req.Method)
+ if req.Method != http.MethodPost {
+ log.Error("Create job, http method not allowed")
+ w.WriteHeader(http.StatusMethodNotAllowed)
+ return
+ }
+ ct := req.Header.Get("Content-Type")
+ if ct != "application/json" {
+ log.Error("Create job, bad content type")
+ http.Error(w, "Bad content type", http.StatusBadRequest)
+ return
+ }
+
+ var t InfoJobDataType
+ err := json.NewDecoder(req.Body).Decode(&t)
+ if err != nil {
+ log.Error("Create job, cannot parse json,", err)
+ http.Error(w, "Cannot parse json", http.StatusBadRequest)
+ return
+ }
+ log.Debug("Creating job, id: ", t.InfoJobIdentity)
+ datalock.Lock()
+ defer datalock.Unlock()
+
+ job_id := t.InfoJobIdentity
+ job_record, job_found := InfoJobs[job_id]
+ type_job_record, found_type := TypeJobs[t.InfoTypeIdentity]
+ if !job_found {
+ if !found_type {
+ log.Error("Type ", t.InfoTypeIdentity, " does not exist")
+ http.Error(w, "Type "+t.InfoTypeIdentity+" does not exist", http.StatusBadRequest)
+ return
+ }
+ } else if t.InfoTypeIdentity != job_record.job_info.InfoTypeIdentity {
+ log.Error("Job cannot change type")
+ http.Error(w, "Job cannot change type", http.StatusBadRequest)
+ return
+ } else if t.InfoJobData.KafkaOutputTopic != job_record.job_info.InfoJobData.KafkaOutputTopic {
+ log.Error("Job cannot change topic")
+ http.Error(w, "Job cannot change topic", http.StatusBadRequest)
+ return
+ } else if !found_type {
+ //Should never happen, if the type is removed then job is stopped
+ log.Error("Type ", t.InfoTypeIdentity, " does not exist")
+ http.Error(w, "Type "+t.InfoTypeIdentity+" does not exist", http.StatusBadRequest)
+ return
+ }
+
+ //TODO: Verify that job contains enough parameters...
+
+ if !job_found {
+ job_record = InfoJobRecord{}
+ job_record.job_info = t
+ output_topic := t.InfoJobData.KafkaOutputTopic
+ job_record.output_topic = t.InfoJobData.KafkaOutputTopic
+ log.Debug("Starting infojob ", job_id, ", type ", t.InfoTypeIdentity, ", output topic", output_topic)
+
+ var stats InfoJobStats
+ job_record.statistics = &stats
+
+ filter := Filter{}
+ filter.JobId = job_id
+ filter.OutputTopic = job_record.output_topic
+
+ jc := JobControl{}
+
+ jc.command = "ADD-FILTER"
+
+ //TODO: Refactor
+ if t.InfoTypeIdentity == "json-file-data-from-filestore" || t.InfoTypeIdentity == "json-file-data" || t.InfoTypeIdentity == "json-file-data-from-filestore-to-influx" {
+ fm := FilterMaps{}
+ fm.sourceNameMap = make(map[string]bool)
+ fm.measObjClassMap = make(map[string]bool)
+ fm.measObjInstIdsMap = make(map[string]bool)
+ fm.measTypesMap = make(map[string]bool)
+ if t.InfoJobData.FilterParams.MeasuredEntityDns != nil {
+ for _, v := range t.InfoJobData.FilterParams.MeasuredEntityDns {
+ fm.sourceNameMap[v] = true
+ }
+ }
+ if t.InfoJobData.FilterParams.MeasObjClass != nil {
+ for _, v := range t.InfoJobData.FilterParams.MeasObjClass {
+ fm.measObjClassMap[v] = true
+ }
+ }
+ if t.InfoJobData.FilterParams.MeasObjInstIds != nil {
+ for _, v := range t.InfoJobData.FilterParams.MeasObjInstIds {
+ fm.measObjInstIdsMap[v] = true
+ }
+ }
+ if t.InfoJobData.FilterParams.MeasTypes != nil {
+ for _, v := range t.InfoJobData.FilterParams.MeasTypes {
+ fm.measTypesMap[v] = true
+ }
+ }
+ filter.filter = fm
+ }
+ if t.InfoTypeIdentity == "json-file-data-from-filestore-to-influx" {
+ influxparam := InfluxJobParameters{}
+ influxparam.DbUrl = t.InfoJobData.DbUrl
+ influxparam.DbOrg = t.InfoJobData.DbOrg
+ influxparam.DbBucket = t.InfoJobData.DbBucket
+ influxparam.DbToken = t.InfoJobData.DbToken
+ filter.influxParameters = influxparam
+ }
+
+ jc.filter = filter
+ InfoJobs[job_id] = job_record
+
+ type_job_record.job_control <- jc
+
+ } else {
+ //TODO
+ //Update job
+ }
+}
+
+// delete job
+func delete_job(w http.ResponseWriter, req *http.Request) {
+ if req.Method != http.MethodDelete {
+ w.WriteHeader(http.StatusMethodNotAllowed)
+ return
+ }
+ datalock.Lock()
+ defer datalock.Unlock()
+
+ vars := mux.Vars(req)
+
+ if id, ok := vars["job_id"]; ok {
+ if _, ok := InfoJobs[id]; ok {
+ remove_info_job(id)
+ w.WriteHeader(http.StatusNoContent)
+ log.Info("Job ", id, " deleted")
+ return
+ }
+ }
+ w.WriteHeader(http.StatusNotFound)
+}
+
+// job supervision
+func supervise_job(w http.ResponseWriter, req *http.Request) {
+ if req.Method != http.MethodGet {
+ w.WriteHeader(http.StatusMethodNotAllowed)
+ return
+ }
+ datalock.Lock()
+ defer datalock.Unlock()
+
+ vars := mux.Vars(req)
+
+ log.Debug("Supervising, job: ", vars["job_id"])
+ if id, ok := vars["job_id"]; ok {
+ if _, ok := InfoJobs[id]; ok {
+ log.Debug("Supervision ok, job", id)
+ return
+ }
+ }
+ w.WriteHeader(http.StatusNotFound)
+}
+
+// producer supervision
+func supervise_producer(w http.ResponseWriter, req *http.Request) {
+ if req.Method != http.MethodGet {
+ w.WriteHeader(http.StatusMethodNotAllowed)
+ return
+ }
+
+ w.WriteHeader(http.StatusOK)
+}
+
+// producer statictics, all jobs
+func statistics(w http.ResponseWriter, req *http.Request) {
+ if req.Method != http.MethodGet {
+ w.WriteHeader(http.StatusMethodNotAllowed)
+ return
+ }
+ m := make(map[string]interface{})
+ log.Debug("producer statictics, locked? ", MutexLocked(&datalock))
+ datalock.Lock()
+ defer datalock.Unlock()
+ req.Header.Set("Content-Type", "application/json; charset=utf-8")
+ m["number-of-jobs"] = len(InfoJobs)
+ m["number-of-types"] = len(InfoTypes.ProdDataTypes)
+ qm := make(map[string]interface{})
+ m["jobs"] = qm
+ for key, elem := range InfoJobs {
+ jm := make(map[string]interface{})
+ qm[key] = jm
+ jm["type"] = elem.job_info.InfoTypeIdentity
+ typeJob := TypeJobs[elem.job_info.InfoTypeIdentity]
+ jm["groupId"] = typeJob.groupId
+ jm["clientID"] = typeJob.clientId
+ jm["input topic"] = typeJob.InputTopic
+ jm["input queue length - job ("+fmt.Sprintf("%v", cap(typeJob.data_in_channel))+")"] = len(typeJob.data_in_channel)
+ jm["output topic"] = elem.output_topic
+ jm["output queue length - producer ("+fmt.Sprintf("%v", cap(data_out_channel))+")"] = len(data_out_channel)
+ jm["msg_in (type)"] = typeJob.statistics.in_msg_cnt
+ jm["msg_out (job)"] = elem.statistics.out_msg_cnt
+
+ }
+ json, err := json.Marshal(m)
+ if err != nil {
+ w.WriteHeader(http.StatusInternalServerError)
+ log.Error("Cannot marshal statistics json")
+ return
+ }
+ _, err = w.Write(json)
+ if err != nil {
+ w.WriteHeader(http.StatusInternalServerError)
+ log.Error("Cannot send statistics json")
+ return
+ }
+}
+
+// Simple alive check
+func alive(w http.ResponseWriter, req *http.Request) {
+ //Alive check
+}
+
+// Get/Set logging level
+func logging_level(w http.ResponseWriter, req *http.Request) {
+ vars := mux.Vars(req)
+ if level, ok := vars["level"]; ok {
+ if req.Method == http.MethodPut {
+ switch level {
+ case "trace":
+ log.SetLevel(log.TraceLevel)
+ case "debug":
+ log.SetLevel(log.DebugLevel)
+ case "info":
+ log.SetLevel(log.InfoLevel)
+ case "warn":
+ log.SetLevel(log.WarnLevel)
+ case "error":
+ log.SetLevel(log.ErrorLevel)
+ case "fatal":
+ log.SetLevel(log.FatalLevel)
+ case "panic":
+ log.SetLevel(log.PanicLevel)
+ default:
+ w.WriteHeader(http.StatusNotFound)
+ }
+ } else {
+ w.WriteHeader(http.StatusMethodNotAllowed)
+ }
+ } else {
+ if req.Method == http.MethodGet {
+ msg := "none"
+ if log.IsLevelEnabled(log.PanicLevel) {
+ msg = "panic"
+ } else if log.IsLevelEnabled(log.FatalLevel) {
+ msg = "fatal"
+ } else if log.IsLevelEnabled(log.ErrorLevel) {
+ msg = "error"
+ } else if log.IsLevelEnabled(log.WarnLevel) {
+ msg = "warn"
+ } else if log.IsLevelEnabled(log.InfoLevel) {
+ msg = "info"
+ } else if log.IsLevelEnabled(log.DebugLevel) {
+ msg = "debug"
+ } else if log.IsLevelEnabled(log.TraceLevel) {
+ msg = "trace"
+ }
+ w.Header().Set("Content-Type", "application/text")
+ w.Write([]byte(msg))
+ } else {
+ w.WriteHeader(http.StatusMethodNotAllowed)
+ }
+ }
+}
+
+func start_topic_reader(topic string, type_id string, control_ch chan ReaderControl, data_ch chan *KafkaPayload, gid string, cid string, stats *TypeJobStats) {
+
+ log.Info("Topic reader starting, topic: ", topic, " for type: ", type_id)
+
+ topic_ok := false
+ var c *kafka.Consumer = nil
+ running := true
+
+ for topic_ok == false {
+
+ select {
+ case reader_ctrl := <-control_ch:
+ if reader_ctrl.command == "EXIT" {
+ log.Info("Topic reader on topic: ", topic, " for type: ", type_id, " - stopped")
+ //TODO: Stop consumer if present?
+ data_ch <- nil //Signal to job handler
+ running = false
+ return
+ }
+ case <-time.After(1 * time.Second):
+ if !running {
+ return
+ }
+ if c == nil {
+ c = create_kafka_consumer(type_id, gid, cid)
+ if c == nil {
+ log.Info("Cannot start consumer on topic: ", topic, " for type: ", type_id, " - retrying")
+ } else {
+ log.Info("Consumer started on topic: ", topic, " for type: ", type_id)
+ }
+ }
+ if c != nil && topic_ok == false {
+ err := c.SubscribeTopics([]string{topic}, nil)
+ if err != nil {
+ log.Info("Topic reader cannot start subscribing on topic: ", topic, " for type: ", type_id, " - retrying -- error details: ", err)
+ } else {
+ log.Info("Topic reader subscribing on topic: ", topic, " for type: ", type_id)
+ topic_ok = true
+ }
+ }
+ }
+ }
+ log.Info("Topic reader ready on topic: ", topic, " for type: ", type_id)
+
+ var event_chan = make(chan int)
+ go func() {
+ for {
+ select {
+ case evt := <-c.Events():
+ switch evt.(type) {
+ case kafka.OAuthBearerTokenRefresh:
+ log.Debug("New consumer token needed: ", evt)
+ token, err := fetch_token()
+ if err != nil {
+ log.Warning("Cannot cannot fetch token: ", err)
+ c.SetOAuthBearerTokenFailure(err.Error())
+ } else {
+ setTokenError := c.SetOAuthBearerToken(*token)
+ if setTokenError != nil {
+ log.Warning("Cannot cannot set token: ", setTokenError)
+ c.SetOAuthBearerTokenFailure(setTokenError.Error())
+ }
+ }
+ default:
+ //TODO: Handle these?
+ log.Debug("Dumping topic reader event on topic: ", topic, " for type: ", type_id, " evt: ", evt.String())
+ }
+
+ case msg := <-event_chan:
+ if msg == 0 {
+ return
+ }
+ case <-time.After(1 * time.Second):
+ if !running {
+ return
+ }
+ }
+ }
+ }()
+
+ go func() {
+ for {
+ //maxDur := 1 * time.Second
+ for {
+ select {
+ case reader_ctrl := <-control_ch:
+ if reader_ctrl.command == "EXIT" {
+ event_chan <- 0
+ log.Debug("Topic reader on topic: ", topic, " for type: ", type_id, " - stopped")
+ data_ch <- nil //Signal to job handler
+ defer c.Close()
+ return
+ }
+ default:
+
+ ev := c.Poll(1000)
+ if ev == nil {
+ log.Debug("Topic Reader for type: ", type_id, " Nothing to consume on topic: ", topic)
+ continue
+ }
+ switch e := ev.(type) {
+ case *kafka.Message:
+ var kmsg KafkaPayload
+ kmsg.msg = e
+
+ c.Commit() //TODO: Ok?
+
+ //TODO: Check for exception
+ data_ch <- &kmsg
+ stats.in_msg_cnt++
+ log.Debug("Reader msg: ", &kmsg)
+ log.Debug("Reader - data_ch ", data_ch)
+ case kafka.Error:
+ fmt.Fprintf(os.Stderr, "%% Error: %v: %v\n", e.Code(), e)
+
+ case kafka.OAuthBearerTokenRefresh:
+ log.Debug("New consumer token needed: ", ev)
+ token, err := fetch_token()
+ if err != nil {
+ log.Warning("Cannot cannot fetch token: ", err)
+ c.SetOAuthBearerTokenFailure(err.Error())
+ } else {
+ setTokenError := c.SetOAuthBearerToken(*token)
+ if setTokenError != nil {
+ log.Warning("Cannot cannot set token: ", setTokenError)
+ c.SetOAuthBearerTokenFailure(setTokenError.Error())
+ }
+ }
+ default:
+ fmt.Printf("Ignored %v\n", e)
+ }
+
+ // orig code
+ // msg, err := c.ReadMessage(maxDur)
+ // if err == nil {
+ // var kmsg KafkaPayload
+ // kmsg.msg = msg
+
+ // c.Commit() //TODO: Ok?
+
+ // //TODO: Check for exception
+ // data_ch <- &kmsg
+ // stats.in_msg_cnt++
+ // log.Debug("Reader msg: ", &kmsg)
+ // log.Debug("Reader - data_ch ", data_ch)
+ // } else {
+ // log.Debug("Topic Reader for type: ", type_id, " Nothing to consume on topic: ", topic, ", reason: ", err)
+ // }
+
+ }
+ }
+ }
+ }()
+}
+
+func start_topic_writer(control_ch chan WriterControl, data_ch chan *KafkaPayload) {
+
+ var kafka_producer *kafka.Producer
+
+ running := true
+ log.Info("Topic writer starting")
+
+ // Wait for kafka producer to become available - and be prepared to exit the writer
+ for kafka_producer == nil {
+ select {
+ case writer_ctl := <-control_ch:
+ if writer_ctl.command == "EXIT" {
+ //ignore cmd
+ }
+ default:
+ kafka_producer = start_producer()
+ if kafka_producer == nil {
+ log.Debug("Could not start kafka producer - retrying")
+ time.Sleep(1 * time.Second)
+ } else {
+ log.Debug("Kafka producer started")
+ //defer kafka_producer.Close()
+ }
+ }
+ }
+
+ var event_chan = make(chan int)
+ go func() {
+ for {
+ select {
+ case evt := <-kafka_producer.Events():
+ //TODO: Handle this? Probably yes, look if the msg was delivered and if not, resend?
+ switch evt.(type) {
+ case *kafka.Message:
+ m := evt.(*kafka.Message)
+
+ if m.TopicPartition.Error != nil {
+ log.Debug("Dumping topic writer event, failed: ", m.TopicPartition.Error)
+ } else {
+ log.Debug("Dumping topic writer event, message to topic: ", *m.TopicPartition.Topic, " at offset: ", m.TopicPartition.Offset, " at partition: ", m.TopicPartition.Partition)
+ }
+ case kafka.Error:
+ log.Debug("Dumping topic writer event, error: ", evt)
+ case kafka.OAuthBearerTokenRefresh:
+ log.Debug("New producer token needed: ", evt)
+ token, err := fetch_token()
+ if err != nil {
+ log.Warning("Cannot cannot fetch token: ", err)
+ kafka_producer.SetOAuthBearerTokenFailure(err.Error())
+ } else {
+ setTokenError := kafka_producer.SetOAuthBearerToken(*token)
+ if setTokenError != nil {
+ log.Warning("Cannot cannot set token: ", setTokenError)
+ kafka_producer.SetOAuthBearerTokenFailure(setTokenError.Error())
+ }
+ }
+ default:
+ log.Debug("Dumping topic writer event, unknown: ", evt)
+ }
+
+ case msg := <-event_chan:
+ if msg == 0 {
+ return
+ }
+ case <-time.After(1 * time.Second):
+ if !running {
+ return
+ }
+ }
+ }
+ }()
+ go func() {
+ for {
+ select {
+ case writer_ctl := <-control_ch:
+ if writer_ctl.command == "EXIT" {
+ // ignore - wait for channel signal
+ }
+
+ case kmsg := <-data_ch:
+ if kmsg == nil {
+ event_chan <- 0
+ // TODO: Close producer?
+ log.Info("Topic writer stopped by channel signal - start_topic_writer")
+ defer kafka_producer.Close()
+ return
+ }
+
+ retries := 10
+ msg_ok := false
+ var err error
+ for retry := 1; retry <= retries && msg_ok == false; retry++ {
+ err = kafka_producer.Produce(&kafka.Message{
+ TopicPartition: kafka.TopicPartition{Topic: &kmsg.topic, Partition: kafka.PartitionAny},
+ Value: kmsg.msg.Value, Key: kmsg.msg.Key}, nil)
+
+ if err == nil {
+ incr_out_msg_cnt(kmsg.jobid)
+ msg_ok = true
+ log.Debug("Topic writer, msg sent ok on topic: ", kmsg.topic)
+ } else {
+ log.Info("Topic writer failed to send message on topic: ", kmsg.topic, " - Retrying. Error details: ", err)
+ time.Sleep(time.Duration(retry) * time.Second)
+ }
+ }
+ if !msg_ok {
+ //TODO: Retry sending msg?
+ log.Error("Topic writer failed to send message on topic: ", kmsg.topic, " - Msg discarded. Error details: ", err)
+ }
+ case <-time.After(1000 * time.Millisecond):
+ if !running {
+ return
+ }
+ }
+ }
+ }()
+}
+
+func create_kafka_consumer(type_id string, gid string, cid string) *kafka.Consumer {
+ var cm kafka.ConfigMap
+ if creds_grant_type == "" {
+ log.Info("Creating kafka SASL plain text consumer for type: ", type_id)
+ cm = kafka.ConfigMap{
+ "bootstrap.servers": bootstrapserver,
+ "group.id": gid,
+ "client.id": cid,
+ "auto.offset.reset": "latest",
+ "enable.auto.commit": false,
+ //"auto.commit.interval.ms": 5000,
+ }
+ } else {
+ log.Info("Creating kafka SASL plain text consumer for type: ", type_id)
+ cm = kafka.ConfigMap{
+ "bootstrap.servers": bootstrapserver,
+ "group.id": gid,
+ "client.id": cid,
+ "auto.offset.reset": "latest",
+ "enable.auto.commit": false,
+ "sasl.mechanism": "OAUTHBEARER",
+ "security.protocol": "SASL_PLAINTEXT",
+ }
+ }
+ c, err := kafka.NewConsumer(&cm)
+
+ //TODO: How to handle autocommit or commit message by message
+ //TODO: Make arg to kafka configurable
+
+ if err != nil {
+ log.Error("Cannot create kafka consumer for type: ", type_id, ", error details: ", err)
+ return nil
+ }
+
+ //c.Commit()
+ log.Info("Created kafka consumer for type: ", type_id, " OK")
+ return c
+}
+
+// Start kafka producer
+func start_producer() *kafka.Producer {
+ log.Info("Creating kafka producer")
+
+ var cm kafka.ConfigMap
+ if creds_grant_type == "" {
+ log.Info("Creating kafka SASL plain text producer")
+ cm = kafka.ConfigMap{
+ "bootstrap.servers": bootstrapserver,
+ }
+ } else {
+ log.Info("Creating kafka SASL plain text producer")
+ cm = kafka.ConfigMap{
+ "bootstrap.servers": bootstrapserver,
+ "sasl.mechanism": "OAUTHBEARER",
+ "security.protocol": "SASL_PLAINTEXT",
+ }
+ }
+
+ p, err := kafka.NewProducer(&cm)
+ if err != nil {
+ log.Error("Cannot create kafka producer,", err)
+ return nil
+ }
+ return p
+}
+
+func start_adminclient() *kafka.AdminClient {
+ log.Info("Creating kafka admin client")
+ a, err := kafka.NewAdminClient(&kafka.ConfigMap{"bootstrap.servers": bootstrapserver})
+ if err != nil {
+ log.Error("Cannot create kafka admin client,", err)
+ return nil
+ }
+ return a
+}
+
+func create_minio_client(id string) (*minio.Client, *error) {
+ log.Debug("Get minio client")
+ minio_client, err := minio.New(filestore_server, &minio.Options{
+ Secure: false,
+ Creds: credentials.NewStaticV4(filestore_user, filestore_pwd, ""),
+ })
+ if err != nil {
+ log.Error("Cannot create minio client, ", err)
+ return nil, &err
+ }
+ return minio_client, nil
+}
+
+func incr_out_msg_cnt(jobid string) {
+ j, ok := InfoJobs[jobid]
+ if ok {
+ j.statistics.out_msg_cnt++
+ }
+}
+
+func start_job_xml_file_data(type_id string, control_ch chan JobControl, data_in_ch chan *KafkaPayload, data_out_channel chan *KafkaPayload, fvolume string, fsbucket string) {
+
+ log.Info("Type job", type_id, " started")
+
+ filters := make(map[string]Filter)
+ topic_list := make(map[string]string)
+ var mc *minio.Client
+ const mc_id = "mc_" + "start_job_xml_file_data"
+ running := true
+ for {
+ select {
+ case job_ctl := <-control_ch:
+ log.Debug("Type job ", type_id, " new cmd received ", job_ctl.command)
+ switch job_ctl.command {
+ case "EXIT":
+ //ignore cmd - handled by channel signal
+ case "ADD-FILTER":
+ filters[job_ctl.filter.JobId] = job_ctl.filter
+ log.Debug("Type job ", type_id, " updated, topic: ", job_ctl.filter.OutputTopic, " jobid: ", job_ctl.filter.JobId)
+
+ tmp_topic_list := make(map[string]string)
+ for k, v := range topic_list {
+ tmp_topic_list[k] = v
+ }
+ tmp_topic_list[job_ctl.filter.JobId] = job_ctl.filter.OutputTopic
+ topic_list = tmp_topic_list
+ case "REMOVE-FILTER":
+ log.Debug("Type job ", type_id, " removing job: ", job_ctl.filter.JobId)
+
+ tmp_topic_list := make(map[string]string)
+ for k, v := range topic_list {
+ tmp_topic_list[k] = v
+ }
+ delete(tmp_topic_list, job_ctl.filter.JobId)
+ topic_list = tmp_topic_list
+ }
+
+ case msg := <-data_in_ch:
+ if msg == nil {
+ log.Info("Type job ", type_id, " stopped by channel signal - start_job_xml_file_data")
+
+ running = false
+ return
+ }
+ if fsbucket != "" && fvolume == "" {
+ if mc == nil {
+ var err *error
+ mc, err = create_minio_client(mc_id)
+ if err != nil {
+ log.Debug("Cannot create minio client for type job: ", type_id)
+ }
+ }
+ }
+ //TODO: Sort processed file conversions in order (FIFO)
+ jobLimiterChan <- struct{}{}
+ go run_xml_job(type_id, msg, "gz", data_out_channel, topic_list, jobLimiterChan, fvolume, fsbucket, mc, mc_id)
+
+ case <-time.After(1 * time.Second):
+ if !running {
+ return
+ }
+ }
+ }
+ //}()
+}
+
+func run_xml_job(type_id string, msg *KafkaPayload, outputCompression string, data_out_channel chan *KafkaPayload, topic_list map[string]string, jobLimiterChan chan struct{}, fvolume string, fsbucket string, mc *minio.Client, mc_id string) {
+ defer func() {
+ <-jobLimiterChan
+ }()
+ PrintMemUsage()
+
+ if fvolume == "" && fsbucket == "" {
+ log.Error("Type job: ", type_id, " cannot run, neither file volume nor filestore is set - discarding message")
+ return
+ } else if (fvolume != "") && (fsbucket != "") {
+ log.Error("Type job: ", type_id, " cannot run with output to both file volume and filestore bucket - discarding message")
+ return
+ }
+
+ start := time.Now()
+ var evt_data XmlFileEventHeader
+
+ err := jsoniter.Unmarshal(msg.msg.Value, &evt_data)
+ if err != nil {
+ log.Error("Cannot parse XmlFileEventHeader for type job: ", type_id, " - discarding message, error details", err)
+ return
+ }
+ log.Debug("Unmarshal file-collect event for type job: ", type_id, " time: ", time.Since(start).String())
+
+ var reader io.Reader
+
+ //TODO -> config
+ INPUTBUCKET := "ropfiles"
+
+ filename := ""
+ if fvolume != "" {
+ filename = fvolume + "/" + evt_data.Name
+ fi, err := os.Open(filename)
+
+ if err != nil {
+ log.Error("File ", filename, " - cannot be opened for type job: ", type_id, " - discarding message, error details: ", err)
+ return
+ }
+ defer fi.Close()
+ reader = fi
+ //} else if evt_data.ObjectStoreBucket != "" {
+ } else {
+ filename = evt_data.Name
+ if mc != nil {
+ tctx := context.Background()
+ mr, err := mc.GetObject(tctx, INPUTBUCKET, filename, minio.GetObjectOptions{})
+ if err != nil {
+ log.Error("Cannot get: ", filename, " for type job: ", type_id, " - discarding message, error details: ", err)
+ return
+ }
+ if mr == nil {
+ log.Error("Cannot get: ", filename, " for type job: ", type_id, " - minio get object returned null reader - discarding message, error details: ", err)
+ return
+ }
+ reader = mr
+ defer mr.Close()
+ } else {
+ log.Error("Cannot get: ", filename, " for type job: ", type_id, " - no minio client - discarding message")
+ return
+ }
+ }
+
+ if reader == nil {
+ log.Error("Cannot get: ", filename, " - null reader")
+ return
+ }
+ var file_bytes []byte
+ if strings.HasSuffix(filename, "gz") {
+ start := time.Now()
+ var buf3 bytes.Buffer
+ errb := gunzipReaderToWriter(&buf3, reader)
+ if errb != nil {
+ log.Error("Cannot gunzip file ", filename, " - discarding message, ", errb)
+ return
+ }
+ file_bytes = buf3.Bytes()
+ log.Debug("Gunzip file: ", filename, "time:", time.Since(start).String(), "len: ", len(file_bytes))
+
+ } else {
+ var buf3 bytes.Buffer
+ _, err2 := io.Copy(&buf3, reader)
+ if err2 != nil {
+ log.Error("File ", filename, " - cannot be read, discarding message, ", err)
+ return
+ }
+ file_bytes = buf3.Bytes()
+ }
+ start = time.Now()
+ b, err := xml_to_json_conv(&file_bytes, &evt_data)
+ if err != nil {
+ log.Error("Cannot convert file ", evt_data.Name, " - discarding message, ", err)
+ return
+ }
+ log.Debug("Converted file to json: ", filename, " time", time.Since(start).String(), "len;", len(b))
+
+ new_fn := evt_data.Name + os.Getenv("KP") + ".json"
+ if outputCompression == "gz" {
+ new_fn = new_fn + ".gz"
+ start = time.Now()
+ var buf bytes.Buffer
+ err = gzipWrite(&buf, &b)
+ if err != nil {
+ log.Error("Cannot gzip file ", new_fn, " - discarding message, ", err)
+ return
+ }
+ b = buf.Bytes()
+ log.Debug("Gzip file: ", new_fn, " time: ", time.Since(start).String(), "len:", len(file_bytes))
+
+ }
+ start = time.Now()
+
+ if fvolume != "" {
+ //Store on disk
+ err = os.WriteFile(fvolume+"/"+new_fn, b, 0644)
+ if err != nil {
+ log.Error("Cannot write file ", new_fn, " - discarding message,", err)
+ return
+ }
+ log.Debug("Write file to disk: "+new_fn, "time: ", time.Since(start).String(), " len: ", len(file_bytes))
+ } else if fsbucket != "" {
+ // Store in minio
+ objectName := new_fn
+ if mc != nil {
+
+ contentType := "application/json"
+ if strings.HasSuffix(objectName, ".gz") {
+ contentType = "application/gzip"
+ }
+
+ // Upload the xml file with PutObject
+ r := bytes.NewReader(b)
+ tctx := context.Background()
+ if check_minio_bucket(mc, mc_id, fsbucket) == false {
+ err := create_minio_bucket(mc, mc_id, fsbucket)
+ if err != nil {
+ log.Error("Cannot create bucket: ", fsbucket, ", ", err)
+ return
+ }
+ }
+ ok := false
+ for i := 1; i < 64 && ok == false; i = i * 2 {
+ info, err := mc.PutObject(tctx, fsbucket, objectName, r, int64(len(b)), minio.PutObjectOptions{ContentType: contentType})
+ if err != nil {
+
+ if i == 1 {
+ log.Warn("Cannot upload (first attempt): ", objectName, ", ", err)
+ } else {
+ log.Warn("Cannot upload (retry): ", objectName, ", ", err)
+ }
+ time.Sleep(time.Duration(i) * time.Second)
+ } else {
+ log.Debug("Store ", objectName, " in filestore, time: ", time.Since(start).String())
+ log.Debug("Successfully uploaded: ", objectName, " of size:", info.Size)
+ ok = true
+ }
+ }
+ if !ok {
+ log.Error("Cannot upload : ", objectName, ", ", err)
+ }
+ } else {
+ log.Error("Cannot upload: ", objectName, ", no client")
+ }
+ }
+
+ start = time.Now()
+ if fvolume == "" {
+ var fde FileDownloadedEvt
+ fde.Filename = new_fn
+ j, err := jsoniter.Marshal(fde)
+
+ if err != nil {
+ log.Error("Cannot marshal FileDownloadedEvt - discarding message, ", err)
+ return
+ }
+ msg.msg.Value = j
+ } else {
+ var fde FileDownloadedEvt
+ fde.Filename = new_fn
+ j, err := jsoniter.Marshal(fde)
+
+ if err != nil {
+ log.Error("Cannot marshal FileDownloadedEvt - discarding message, ", err)
+ return
+ }
+ msg.msg.Value = j
+ }
+ msg.msg.Key = []byte("\"" + evt_data.SourceName + "\"")
+ log.Debug("Marshal file-collect event ", time.Since(start).String())
+
+ for k, v := range topic_list {
+ var kmsg *KafkaPayload = new(KafkaPayload)
+ kmsg.msg = msg.msg
+ kmsg.topic = v
+ kmsg.jobid = k
+ data_out_channel <- kmsg
+ }
+}
+
+func xml_to_json_conv(f_byteValue *[]byte, xfeh *XmlFileEventHeader) ([]byte, error) {
+ var f MeasCollecFile
+ start := time.Now()
+ err := xml.Unmarshal(*f_byteValue, &f)
+ if err != nil {
+ return nil, errors.New("Cannot unmarshal xml-file")
+ }
+ log.Debug("Unmarshal xml file XmlFileEvent: ", time.Since(start).String())
+
+ start = time.Now()
+ var pmfile PMJsonFile
+
+ //TODO: Fill in more values
+ pmfile.Event.Perf3GppFields.Perf3GppFieldsVersion = "1.0"
+ pmfile.Event.Perf3GppFields.MeasDataCollection.GranularityPeriod = 900
+ pmfile.Event.Perf3GppFields.MeasDataCollection.MeasuredEntityUserName = ""
+ pmfile.Event.Perf3GppFields.MeasDataCollection.MeasuredEntityDn = f.FileHeader.FileSender.LocalDn
+ pmfile.Event.Perf3GppFields.MeasDataCollection.MeasuredEntitySoftwareVersion = f.MeasData.ManagedElement.SwVersion
+
+ for _, it := range f.MeasData.MeasInfo {
+ var mili MeasInfoList
+ mili.MeasInfoID.SMeasInfoID = it.MeasInfoId
+ for _, jt := range it.MeasType {
+ mili.MeasTypes.SMeasTypesList = append(mili.MeasTypes.SMeasTypesList, jt.Text)
+ }
+ for _, jt := range it.MeasValue {
+ var mv MeasValues
+ mv.MeasObjInstID = jt.MeasObjLdn
+ mv.SuspectFlag = jt.Suspect
+ if jt.Suspect == "" {
+ mv.SuspectFlag = "false"
+ }
+ for _, kt := range jt.R {
+ ni, _ := strconv.Atoi(kt.P)
+ nv := kt.Text
+ mr := MeasResults{ni, nv}
+ mv.MeasResultsList = append(mv.MeasResultsList, mr)
+ }
+ mili.MeasValuesList = append(mili.MeasValuesList, mv)
+ }
+
+ pmfile.Event.Perf3GppFields.MeasDataCollection.SMeasInfoList = append(pmfile.Event.Perf3GppFields.MeasDataCollection.SMeasInfoList, mili)
+ }
+
+ pmfile.Event.Perf3GppFields.MeasDataCollection.GranularityPeriod = 900
+
+ //TODO: Fill more values
+ pmfile.Event.CommonEventHeader.Domain = "" //xfeh.Domain
+ pmfile.Event.CommonEventHeader.EventID = "" //xfeh.EventID
+ pmfile.Event.CommonEventHeader.Sequence = 0 //xfeh.Sequence
+ pmfile.Event.CommonEventHeader.EventName = "" //xfeh.EventName
+ pmfile.Event.CommonEventHeader.SourceName = xfeh.SourceName
+ pmfile.Event.CommonEventHeader.ReportingEntityName = "" //xfeh.ReportingEntityName
+ pmfile.Event.CommonEventHeader.Priority = "" //xfeh.Priority
+ pmfile.Event.CommonEventHeader.StartEpochMicrosec = xfeh.StartEpochMicrosec
+ pmfile.Event.CommonEventHeader.LastEpochMicrosec = xfeh.LastEpochMicrosec
+ pmfile.Event.CommonEventHeader.Version = "" //xfeh.Version
+ pmfile.Event.CommonEventHeader.VesEventListenerVersion = "" //xfeh.VesEventListenerVersion
+ pmfile.Event.CommonEventHeader.TimeZoneOffset = xfeh.TimeZoneOffset
+
+ log.Debug("Convert xml to json : ", time.Since(start).String())
+
+ start = time.Now()
+ json, err := jsoniter.Marshal(pmfile)
+ log.Debug("Marshal json : ", time.Since(start).String())
+
+ if err != nil {
+ return nil, errors.New("Cannot marshal converted json")
+ }
+ return json, nil
+}
+
+func start_job_json_file_data(type_id string, control_ch chan JobControl, data_in_ch chan *KafkaPayload, data_out_channel chan *KafkaPayload, objectstore bool) {
+
+ log.Info("Type job", type_id, " started")
+
+ filters := make(map[string]Filter)
+ filterParams_list := make(map[string]FilterMaps)
+ // ch_list := make(map[string]chan *KafkaPayload)
+ topic_list := make(map[string]string)
+ var mc *minio.Client
+ const mc_id = "mc_" + "start_job_json_file_data"
+ running := true
+ for {
+ select {
+ case job_ctl := <-control_ch:
+ log.Debug("Type job ", type_id, " new cmd received ", job_ctl.command)
+ switch job_ctl.command {
+ case "EXIT":
+ //ignore cmd - handled by channel signal
+ case "ADD-FILTER":
+ //TODO: Refactor...
+ filters[job_ctl.filter.JobId] = job_ctl.filter
+ log.Debug("Type job ", type_id, " updated, topic: ", job_ctl.filter.OutputTopic, " jobid: ", job_ctl.filter.JobId)
+
+ tmp_filterParams_list := make(map[string]FilterMaps)
+ for k, v := range filterParams_list {
+ tmp_filterParams_list[k] = v
+ }
+ tmp_filterParams_list[job_ctl.filter.JobId] = job_ctl.filter.filter
+ filterParams_list = tmp_filterParams_list
+
+ tmp_topic_list := make(map[string]string)
+ for k, v := range topic_list {
+ tmp_topic_list[k] = v
+ }
+ tmp_topic_list[job_ctl.filter.JobId] = job_ctl.filter.OutputTopic
+ topic_list = tmp_topic_list
+ case "REMOVE-FILTER":
+ //TODO: Refactor...
+ log.Debug("Type job ", type_id, " removing job: ", job_ctl.filter.JobId)
+
+ tmp_filterParams_list := make(map[string]FilterMaps)
+ for k, v := range filterParams_list {
+ tmp_filterParams_list[k] = v
+ }
+ delete(tmp_filterParams_list, job_ctl.filter.JobId)
+ filterParams_list = tmp_filterParams_list
+
+ tmp_topic_list := make(map[string]string)
+ for k, v := range topic_list {
+ tmp_topic_list[k] = v
+ }
+ delete(tmp_topic_list, job_ctl.filter.JobId)
+ topic_list = tmp_topic_list
+ }
+
+ case msg := <-data_in_ch:
+ if msg == nil {
+ log.Info("Type job ", type_id, " stopped by channel signal - start_job_xml_file_data")
+
+ running = false
+ return
+ }
+ if objectstore {
+ if mc == nil {
+ var err *error
+ mc, err = create_minio_client(mc_id)
+ if err != nil {
+ log.Debug("Cannot create minio client for type job: ", type_id)
+ }
+ }
+ }
+ //TODO: Sort processed file conversions in order (FIFO)
+ jobLimiterChan <- struct{}{}
+ go run_json_job(type_id, msg, "", filterParams_list, topic_list, data_out_channel, jobLimiterChan, mc, mc_id, objectstore)
+
+ case <-time.After(1 * time.Second):
+ if !running {
+ return
+ }
+ }
+ }
+}
+
+func run_json_job(type_id string, msg *KafkaPayload, outputCompression string, filterList map[string]FilterMaps, topic_list map[string]string, data_out_channel chan *KafkaPayload, jobLimiterChan chan struct{}, mc *minio.Client, mc_id string, objectstore bool) {
+
+ //Release job limit
+ defer func() {
+ <-jobLimiterChan
+ }()
+
+ PrintMemUsage()
+
+ var evt_data FileDownloadedEvt
+ err := jsoniter.Unmarshal(msg.msg.Value, &evt_data)
+ if err != nil {
+ log.Error("Cannot parse FileDownloadedEvt for type job: ", type_id, " - discarding message, error details: ", err)
+ return
+ }
+ log.Debug("FileDownloadedEvt for job: ", type_id, " file: ", evt_data.Filename)
+
+ var reader io.Reader
+
+ //TODO -> config
+ //INPUTBUCKET := "json-file-ready"
+ INPUTBUCKET := "pm-files-json"
+ filename := ""
+ if objectstore == false {
+ filename = files_volume + "/" + evt_data.Filename
+ fi, err := os.Open(filename)
+
+ if err != nil {
+ log.Error("File: ", filename, "for type job: ", type_id, " - cannot be opened -discarding message - error details: ", err)
+ return
+ }
+ defer fi.Close()
+ reader = fi
+ } else {
+ filename = "/" + evt_data.Filename
+ if mc != nil {
+ if check_minio_bucket(mc, mc_id, INPUTBUCKET) == false {
+ log.Error("Bucket not available for reading in type job: ", type_id, "bucket: ", INPUTBUCKET)
+ return
+ }
+ tctx := context.Background()
+ mr, err := mc.GetObject(tctx, INPUTBUCKET, filename, minio.GetObjectOptions{})
+ if err != nil {
+ log.Error("Obj: ", filename, "for type job: ", type_id, " - cannot be fetched from bucket: ", INPUTBUCKET, " - discarding message, error details: ", err)
+ return
+ }
+ reader = mr
+ defer mr.Close()
+ } else {
+ log.Error("Cannot get obj: ", filename, "for type job: ", type_id, " from bucket: ", INPUTBUCKET, " - no client")
+ return
+ }
+ }
+
+ var data *[]byte
+ if strings.HasSuffix(filename, "gz") {
+ start := time.Now()
+ var buf2 bytes.Buffer
+ errb := gunzipReaderToWriter(&buf2, reader)
+ if errb != nil {
+ log.Error("Cannot decompress file/obj ", filename, "for type job: ", type_id, " - discarding message, error details", errb)
+ return
+ }
+ d := buf2.Bytes()
+ data = &d
+ log.Debug("Decompress file/obj ", filename, "for type job: ", type_id, " time:", time.Since(start).String())
+ } else {
+
+ start := time.Now()
+ d, err := io.ReadAll(reader)
+ if err != nil {
+ log.Error("Cannot read file/obj: ", filename, "for type job: ", type_id, " - discarding message, error details", err)
+ return
+ }
+ data = &d
+
+ log.Debug("Read file/obj: ", filename, "for type job: ", type_id, " time: ", time.Since(start).String())
+ }
+
+ // The code was moved to inside the below for loop - a deep copy of PM PMJsonFile is need to keep it outside
+ // var pmfile PMJsonFile
+ // start := time.Now()
+ // err = jsoniter.Unmarshal(*data, &pmfile)
+ // log.Debug("Json unmarshal obj: ", filename, " time: ", time.Since(start).String())
+
+ // if err != nil {
+ // log.Error("Msg could not be unmarshalled - discarding message, obj: ", filename, " - error details:", err)
+ // return
+ // }
+ for k, v := range filterList {
+
+ var pmfile PMJsonFile
+ start := time.Now()
+ err = jsoniter.Unmarshal(*data, &pmfile)
+ log.Debug("Json unmarshal obj: ", filename, " time: ", time.Since(start).String())
+
+ if err != nil {
+ log.Error("Msg could not be unmarshalled - discarding message, obj: ", filename, " - error details:", err)
+ return
+ }
+
+ var kmsg *KafkaPayload = new(KafkaPayload)
+ kmsg.msg = new(kafka.Message)
+ kmsg.msg.Key = []byte("\"" + pmfile.Event.CommonEventHeader.SourceName + "\"")
+ log.Debug("topic:", topic_list[k])
+ log.Debug("sourceNameMap:", v.sourceNameMap)
+ log.Debug("measObjClassMap:", v.measObjClassMap)
+ log.Debug("measObjInstIdsMap:", v.measObjInstIdsMap)
+ log.Debug("measTypesMap:", v.measTypesMap)
+ //BMX if len(v.sourceNameMap) > 0 || len(v.measObjInstIdsMap) > 0 || len(v.measTypesMap) > 0 || len(v.measObjClassMap) > 0 {
+ b := json_pm_filter_to_byte(evt_data.Filename, &pmfile, v.sourceNameMap, v.measObjClassMap, v.measObjInstIdsMap, v.measTypesMap)
+ if b == nil {
+ log.Info("File/obj ", filename, "for job: ", k, " - empty after filering, discarding message ")
+ return
+ }
+ kmsg.msg.Value = *b
+ //BMX}
+
+ // if outputCompression == "json.gz" {
+ // start := time.Now()
+ // var buf bytes.Buffer
+ // err := gzipWrite(&buf, &kmsg.msg.Value)
+ // if err != nil {
+ // log.Error("Cannot compress file/obj ", filename, "for job: ", job_id, " - discarding message, error details", err)
+ // return
+
+ // }
+ // kmsg.msg.Value = buf.Bytes()
+ // log.Debug("Compress file/obj ", filename, "for job: ", job_id, " time:", time.Since(start).String())
+ // }
+ kmsg.topic = topic_list[k]
+ kmsg.jobid = k
+
+ data_out_channel <- kmsg
+ }
+
+}
+
+func json_pm_filter_to_byte(resource string, data *PMJsonFile, sourceNameMap map[string]bool, measObjClassMap map[string]bool, measObjInstIdsMap map[string]bool, measTypesMap map[string]bool) *[]byte {
+
+ if json_pm_filter_to_obj(resource, data, sourceNameMap, measObjClassMap, measObjInstIdsMap, measTypesMap) == nil {
+ return nil
+ }
+ start := time.Now()
+ j, err := jsoniter.Marshal(&data)
+
+ log.Debug("Json marshal obj: ", resource, " time: ", time.Since(start).String())
+
+ if err != nil {
+ log.Error("Msg could not be marshalled discarding message, obj: ", resource, ", error details: ", err)
+ return nil
+ }
+
+ log.Debug("Filtered json obj: ", resource, " len: ", len(j))
+ return &j
+}
+
+func json_pm_filter_to_obj(resource string, data *PMJsonFile, sourceNameMap map[string]bool, measObjClassMap map[string]bool, measObjInstIdsMap map[string]bool, measTypesMap map[string]bool) *PMJsonFile {
+ filter_req := true
+ start := time.Now()
+ if len(sourceNameMap) != 0 {
+ if !sourceNameMap[data.Event.CommonEventHeader.SourceName] {
+ filter_req = false
+ return nil
+ }
+ }
+ if filter_req {
+ modified := false
+ var temp_mil []MeasInfoList
+ for _, zz := range data.Event.Perf3GppFields.MeasDataCollection.SMeasInfoList {
+
+ check_cntr := false
+ var cnt_flags []bool
+ if len(measTypesMap) > 0 {
+ c_cntr := 0
+ var temp_mtl []string
+ for _, v := range zz.MeasTypes.SMeasTypesList {
+ if measTypesMap[v] {
+ cnt_flags = append(cnt_flags, true)
+ c_cntr++
+ temp_mtl = append(temp_mtl, v)
+ } else {
+ cnt_flags = append(cnt_flags, false)
+ }
+ }
+ if c_cntr > 0 {
+ check_cntr = true
+ zz.MeasTypes.SMeasTypesList = temp_mtl
+ } else {
+ modified = true
+ continue
+ }
+ }
+ keep := false
+ var temp_mvl []MeasValues
+ for _, yy := range zz.MeasValuesList {
+ keep_class := false
+ keep_inst := false
+ keep_cntr := false
+
+ dna := strings.Split(yy.MeasObjInstID, ",")
+ instName := dna[len(dna)-1]
+ cls := strings.Split(dna[len(dna)-1], "=")[0]
+
+ if len(measObjClassMap) > 0 {
+ if measObjClassMap[cls] {
+ keep_class = true
+ }
+ } else {
+ keep_class = true
+ }
+
+ if len(measObjInstIdsMap) > 0 {
+ if measObjInstIdsMap[instName] {
+ keep_inst = true
+ }
+ } else {
+ keep_inst = true
+ }
+
+ if check_cntr {
+ var temp_mrl []MeasResults
+ cnt_p := 1
+ for _, v := range yy.MeasResultsList {
+ if cnt_flags[v.P-1] {
+ v.P = cnt_p
+ cnt_p++
+ temp_mrl = append(temp_mrl, v)
+ }
+ }
+ yy.MeasResultsList = temp_mrl
+ keep_cntr = true
+ } else {
+ keep_cntr = true
+ }
+ if keep_class && keep_cntr && keep_inst {
+ keep = true
+ temp_mvl = append(temp_mvl, yy)
+ }
+ }
+ if keep {
+ zz.MeasValuesList = temp_mvl
+ temp_mil = append(temp_mil, zz)
+ modified = true
+ }
+
+ }
+ //Only if modified
+ if modified {
+ if len(temp_mil) == 0 {
+ log.Debug("Msg filtered, nothing found, discarding, obj: ", resource)
+ return nil
+ }
+ data.Event.Perf3GppFields.MeasDataCollection.SMeasInfoList = temp_mil
+ }
+ }
+ log.Debug("Filter: ", time.Since(start).String())
+ return data
+}
+
+// func json_pm_filter(resource string, data *PMJsonFile, sourceNameMap map[string]bool, measObjClassMap map[string]bool, measObjInstIdsMap map[string]bool, measTypesMap map[string]bool) *[]byte {
+
+// filter_req := true
+// start := time.Now()
+// if len(sourceNameMap) != 0 {
+// if !sourceNameMap[data.Event.CommonEventHeader.SourceName] {
+// filter_req = false
+// return nil
+// }
+// }
+// if filter_req {
+// modified := false
+// var temp_mil []MeasInfoList
+// for _, zz := range data.Event.Perf3GppFields.MeasDataCollection.SMeasInfoList {
+
+// check_cntr := false
+// var cnt_flags []bool
+// if len(measTypesMap) > 0 {
+// c_cntr := 0
+// var temp_mtl []string
+// for _, v := range zz.MeasTypes.SMeasTypesList {
+// if measTypesMap[v] {
+// cnt_flags = append(cnt_flags, true)
+// c_cntr++
+// temp_mtl = append(temp_mtl, v)
+// } else {
+// cnt_flags = append(cnt_flags, false)
+// }
+// }
+// if c_cntr > 0 {
+// check_cntr = true
+// zz.MeasTypes.SMeasTypesList = temp_mtl
+// } else {
+// modified = true
+// continue
+// }
+// }
+// keep := false
+// var temp_mvl []MeasValues
+// for _, yy := range zz.MeasValuesList {
+// keep_class := false
+// keep_inst := false
+// keep_cntr := false
+
+// dna := strings.Split(yy.MeasObjInstID, ",")
+// instName := dna[len(dna)-1]
+// cls := strings.Split(dna[len(dna)-1], "=")[0]
+
+// if len(measObjClassMap) > 0 {
+// if measObjClassMap[cls] {
+// keep_class = true
+// }
+// } else {
+// keep_class = true
+// }
+
+// if len(measObjInstIdsMap) > 0 {
+// if measObjInstIdsMap[instName] {
+// keep_inst = true
+// }
+// } else {
+// keep_inst = true
+// }
+
+// if check_cntr {
+// var temp_mrl []MeasResults
+// cnt_p := 1
+// for _, v := range yy.MeasResultsList {
+// if cnt_flags[v.P-1] {
+// v.P = cnt_p
+// cnt_p++
+// temp_mrl = append(temp_mrl, v)
+// }
+// }
+// yy.MeasResultsList = temp_mrl
+// keep_cntr = true
+// } else {
+// keep_cntr = true
+// }
+// if keep_class && keep_cntr && keep_inst {
+// keep = true
+// temp_mvl = append(temp_mvl, yy)
+// }
+// }
+// if keep {
+// zz.MeasValuesList = temp_mvl
+// temp_mil = append(temp_mil, zz)
+// modified = true
+// }
+
+// }
+// //Only if modified
+// if modified {
+// if len(temp_mil) == 0 {
+// log.Debug("Msg filtered, nothing found, discarding, obj: ", resource)
+// return nil
+// }
+// data.Event.Perf3GppFields.MeasDataCollection.SMeasInfoList = temp_mil
+// }
+// }
+// log.Debug("Filter: ", time.Since(start).String())
+
+// start = time.Now()
+// j, err := jsoniter.Marshal(&data)
+
+// log.Debug("Json marshal obj: ", resource, " time: ", time.Since(start).String())
+
+// if err != nil {
+// log.Error("Msg could not be marshalled discarding message, obj: ", resource, ", error details: ", err)
+// return nil
+// }
+
+// log.Debug("Filtered json obj: ", resource, " len: ", len(j))
+// return &j
+// }
+
+func start_job_json_file_data_influx(type_id string, control_ch chan JobControl, data_in_ch chan *KafkaPayload, objectstore bool) {
+
+ log.Info("Type job", type_id, " started")
+ log.Debug("influx job ch ", data_in_ch)
+ filters := make(map[string]Filter)
+ filterParams_list := make(map[string]FilterMaps)
+ influx_job_params := make(map[string]InfluxJobParameters)
+ var mc *minio.Client
+ const mc_id = "mc_" + "start_job_json_file_data_influx"
+ running := true
+ for {
+ select {
+ case job_ctl := <-control_ch:
+ log.Debug("Type job ", type_id, " new cmd received ", job_ctl.command)
+ switch job_ctl.command {
+ case "EXIT":
+ //ignore cmd - handled by channel signal
+ case "ADD-FILTER":
+ //TODO: Refactor...
+ filters[job_ctl.filter.JobId] = job_ctl.filter
+ log.Debug("Type job ", type_id, " updated, topic: ", job_ctl.filter.OutputTopic, " jobid: ", job_ctl.filter.JobId)
+ log.Debug(job_ctl.filter)
+ tmp_filterParams_list := make(map[string]FilterMaps)
+ for k, v := range filterParams_list {
+ tmp_filterParams_list[k] = v
+ }
+ tmp_filterParams_list[job_ctl.filter.JobId] = job_ctl.filter.filter
+ filterParams_list = tmp_filterParams_list
+
+ tmp_influx_job_params := make(map[string]InfluxJobParameters)
+ for k, v := range influx_job_params {
+ tmp_influx_job_params[k] = v
+ }
+ tmp_influx_job_params[job_ctl.filter.JobId] = job_ctl.filter.influxParameters
+ influx_job_params = tmp_influx_job_params
+
+ case "REMOVE-FILTER":
+ //TODO: Refactor...
+ log.Debug("Type job ", type_id, " removing job: ", job_ctl.filter.JobId)
+
+ tmp_filterParams_list := make(map[string]FilterMaps)
+ for k, v := range filterParams_list {
+ tmp_filterParams_list[k] = v
+ }
+ delete(tmp_filterParams_list, job_ctl.filter.JobId)
+ filterParams_list = tmp_filterParams_list
+
+ tmp_influx_job_params := make(map[string]InfluxJobParameters)
+ for k, v := range influx_job_params {
+ tmp_influx_job_params[k] = v
+ }
+ delete(tmp_influx_job_params, job_ctl.filter.JobId)
+ influx_job_params = tmp_influx_job_params
+ }
+
+ case msg := <-data_in_ch:
+ log.Debug("Data reveived - influx")
+ if msg == nil {
+ log.Info("Type job ", type_id, " stopped by channel signal - start_job_xml_file_data")
+
+ running = false
+ return
+ }
+ if objectstore {
+ if mc == nil {
+ var err *error
+ mc, err = create_minio_client(mc_id)
+ if err != nil {
+ log.Debug("Cannot create minio client for type job: ", type_id)
+ }
+ }
+ }
+ //TODO: Sort processed file conversions in order (FIFO)
+ jobLimiterChan <- struct{}{}
+ go run_json_file_data_job_influx(type_id, msg, filterParams_list, influx_job_params, jobLimiterChan, mc, mc_id, objectstore)
+
+ case <-time.After(1 * time.Second):
+ if !running {
+ return
+ }
+ }
+ }
+}
+
+func run_json_file_data_job_influx(type_id string, msg *KafkaPayload, filterList map[string]FilterMaps, influxList map[string]InfluxJobParameters, jobLimiterChan chan struct{}, mc *minio.Client, mc_id string, objectstore bool) {
+
+ log.Debug("run_json_file_data_job_influx")
+ //Release job limit
+ defer func() {
+ <-jobLimiterChan
+ }()
+
+ PrintMemUsage()
+
+ var evt_data FileDownloadedEvt
+ err := jsoniter.Unmarshal(msg.msg.Value, &evt_data)
+ if err != nil {
+ log.Error("Cannot parse FileDownloadedEvt for type job: ", type_id, " - discarding message, error details: ", err)
+ return
+ }
+ log.Debug("FileDownloadedEvt for job: ", type_id, " file: ", evt_data.Filename)
+
+ var reader io.Reader
+
+ //TODO -> config
+ //INPUTBUCKET := "json-file-ready"
+ INPUTBUCKET := "pm-files-json"
+ filename := ""
+ if objectstore == false {
+ filename = files_volume + "/" + evt_data.Filename
+ fi, err := os.Open(filename)
+
+ if err != nil {
+ log.Error("File: ", filename, "for type job: ", type_id, " - cannot be opened -discarding message - error details: ", err)
+ return
+ }
+ defer fi.Close()
+ reader = fi
+ } else {
+ filename = "/" + evt_data.Filename
+ if mc != nil {
+ if check_minio_bucket(mc, mc_id, INPUTBUCKET) == false {
+ log.Error("Bucket not available for reading in type job: ", type_id, "bucket: ", INPUTBUCKET)
+ return
+ }
+ tctx := context.Background()
+ mr, err := mc.GetObject(tctx, INPUTBUCKET, filename, minio.GetObjectOptions{})
+ if err != nil {
+ log.Error("Obj: ", filename, "for type job: ", type_id, " - cannot be fetched from bucket: ", INPUTBUCKET, " - discarding message, error details: ", err)
+ return
+ }
+ reader = mr
+ defer mr.Close()
+ } else {
+ log.Error("Cannot get obj: ", filename, "for type job: ", type_id, " from bucket: ", INPUTBUCKET, " - no client")
+ return
+ }
+ }
+
+ var data *[]byte
+ if strings.HasSuffix(filename, "gz") {
+ start := time.Now()
+ var buf2 bytes.Buffer
+ errb := gunzipReaderToWriter(&buf2, reader)
+ if errb != nil {
+ log.Error("Cannot decompress file/obj ", filename, "for type job: ", type_id, " - discarding message, error details", errb)
+ return
+ }
+ d := buf2.Bytes()
+ data = &d
+ log.Debug("Decompress file/obj ", filename, "for type job: ", type_id, " time:", time.Since(start).String())
+ } else {
+
+ start := time.Now()
+ d, err := io.ReadAll(reader)
+ if err != nil {
+ log.Error("Cannot read file/obj: ", filename, "for type job: ", type_id, " - discarding message, error details", err)
+ return
+ }
+ data = &d
+
+ log.Debug("Read file/obj: ", filename, "for type job: ", type_id, " time: ", time.Since(start).String())
+ }
+ // The code was moved to inside the below for loop - a deep copy of PM PMJsonFile is need to keep it outside
+ // var pmfile PMJsonFile
+ // start := time.Now()
+ // err = jsoniter.Unmarshal(*data, &pmfile)
+ // log.Debug("Json unmarshal obj: ", filename, " time: ", time.Since(start).String())
+
+ // if err != nil {
+ // log.Error("Msg could not be unmarshalled - discarding message, obj: ", filename, " - error details:", err)
+ // return
+ // }
+ for k, v := range filterList {
+
+ var pmfile PMJsonFile
+ start := time.Now()
+ err = jsoniter.Unmarshal(*data, &pmfile)
+ log.Debug("Json unmarshal obj: ", filename, " time: ", time.Since(start).String())
+
+ if err != nil {
+ log.Error("Msg could not be unmarshalled - discarding message, obj: ", filename, " - error details:", err)
+ return
+ }
+
+ if len(v.sourceNameMap) > 0 || len(v.measObjInstIdsMap) > 0 || len(v.measTypesMap) > 0 || len(v.measObjClassMap) > 0 {
+ b := json_pm_filter_to_obj(evt_data.Filename, &pmfile, v.sourceNameMap, v.measObjClassMap, v.measObjInstIdsMap, v.measTypesMap)
+ if b == nil {
+ log.Info("File/obj ", filename, "for job: ", k, " - empty after filering, discarding message ")
+ return
+ }
+
+ }
+ fluxParms := influxList[k]
+ log.Debug("Influxdb params: ", fluxParms)
+ client := influxdb2.NewClient(fluxParms.DbUrl, fluxParms.DbToken)
+ writeAPI := client.WriteAPIBlocking(fluxParms.DbOrg, fluxParms.DbBucket)
+
+ // fmt.Println(pmfile.Event.CommonEventHeader.StartEpochMicrosec)
+ // tUnix := pmfile.Event.CommonEventHeader.StartEpochMicrosec / int64(time.Millisecond)
+ // tUnixNanoRemainder := (pmfile.Event.CommonEventHeader.StartEpochMicrosec % int64(time.Millisecond)) * int64(time.Microsecond)
+ // timeT := time.Unix(tUnix, tUnixNanoRemainder)
+ // fmt.Println(timeT)
+ // fmt.Println("======================")
+ for _, zz := range pmfile.Event.Perf3GppFields.MeasDataCollection.SMeasInfoList {
+ ctr_names := make(map[string]string)
+ for cni, cn := range zz.MeasTypes.SMeasTypesList {
+ ctr_names[string(cni+1)] = cn
+ }
+ for _, xx := range zz.MeasValuesList {
+ log.Debug("Measurement: ", xx.MeasObjInstID)
+ log.Debug("Suspect flag: ", xx.SuspectFlag)
+ p := influxdb2.NewPointWithMeasurement(xx.MeasObjInstID)
+ p.AddField("suspectflag", xx.SuspectFlag)
+ p.AddField("granularityperiod", pmfile.Event.Perf3GppFields.MeasDataCollection.GranularityPeriod)
+ p.AddField("timezone", pmfile.Event.CommonEventHeader.TimeZoneOffset)
+ for _, yy := range xx.MeasResultsList {
+ pi := string(yy.P)
+ pv := yy.SValue
+ pn := ctr_names[pi]
+ log.Debug("Counter: ", pn, " Value: ", pv)
+ pv_i, err := strconv.Atoi(pv)
+ if err == nil {
+ p.AddField(pn, pv_i)
+ } else {
+ p.AddField(pn, pv)
+ }
+ }
+ //p.SetTime(timeT)
+ log.Debug("StartEpochMicrosec from common event header: ", pmfile.Event.CommonEventHeader.StartEpochMicrosec)
+ log.Debug("Set time: ", time.Unix(pmfile.Event.CommonEventHeader.StartEpochMicrosec/1000000, 0))
+ p.SetTime(time.Unix(pmfile.Event.CommonEventHeader.StartEpochMicrosec/1000000, 0))
+ err := writeAPI.WritePoint(context.Background(), p)
+ if err != nil {
+ log.Error("Db write error: ", err)
+ }
+ }
+
+ }
+ client.Close()
+ }
+
+}
Code Review / nonrtric / plt / ranpm.git / blobdiff