dmaap-mediator-producer/internal/jobs/jobs.go

   1 // -
   2 //   ========================LICENSE_START=================================
   3 //   O-RAN-SC
   4 //   %%
   5 //   Copyright (C) 2021: Nordix Foundation
   6 //   %%
   7 //   Licensed under the Apache License, Version 2.0 (the "License");
   8 //   you may not use this file except in compliance with the License.
   9 //   You may obtain a copy of the License at
  10 //
  11 //        http://www.apache.org/licenses/LICENSE-2.0
  12 //
  13 //   Unless required by applicable law or agreed to in writing, software
  14 //   distributed under the License is distributed on an "AS IS" BASIS,
  15 //   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 //   See the License for the specific language governing permissions and
  17 //   limitations under the License.
  18 //   ========================LICENSE_END===================================
  19 //
  20
  21 package jobs
  22
  23 import (
  24         "encoding/json"
  25         "fmt"
  26         "os"
  27         "sync"
  28
  29         log "github.com/sirupsen/logrus"
  30         "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
  31         "oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient"
  32 )
  33
  34 type TypeData struct {
  35         TypeId        string `json:"id"`
  36         DMaaPTopicURL string `json:"dmaapTopicUrl"`
  37         jobHandler    *jobHandler
  38 }
  39
  40 type JobInfo struct {
  41         Owner            string      `json:"owner"`
  42         LastUpdated      string      `json:"last_updated"`
  43         InfoJobIdentity  string      `json:"info_job_identity"`
  44         TargetUri        string      `json:"target_uri"`
  45         InfoJobData      interface{} `json:"info_job_data"`
  46         InfoTypeIdentity string      `json:"info_type_identity"`
  47 }
  48
  49 type JobTypesManager interface {
  50         LoadTypesFromConfiguration() ([]config.TypeDefinition, error)
  51         GetSupportedTypes() []string
  52 }
  53
  54 type JobsManager interface {
  55         AddJob(JobInfo) error
  56         DeleteJob(jobId string)
  57 }
  58
  59 type JobsManagerImpl struct {
  60         configFile       string
  61         allTypes         map[string]TypeData
  62         pollClient       restclient.HTTPClient
  63         mrAddress        string
  64         distributeClient restclient.HTTPClient
  65 }
  66
  67 type jobHandler struct {
  68         mu               sync.Mutex
  69         typeId           string
  70         topicUrl         string
  71         jobs             map[string]JobInfo
  72         addJobCh         chan JobInfo
  73         deleteJobCh      chan string
  74         pollClient       restclient.HTTPClient
  75         distributeClient restclient.HTTPClient
  76 }
  77
  78 func NewJobsManagerImpl(typeConfigFilePath string, pollClient restclient.HTTPClient, mrAddr string, distributeClient restclient.HTTPClient) *JobsManagerImpl {
  79         return &JobsManagerImpl{
  80                 configFile:       typeConfigFilePath,
  81                 allTypes:         make(map[string]TypeData),
  82                 pollClient:       pollClient,
  83                 mrAddress:        mrAddr,
  84                 distributeClient: distributeClient,
  85         }
  86 }
  87
  88 func (jm *JobsManagerImpl) AddJob(ji JobInfo) error {
  89         if err := jm.validateJobInfo(ji); err == nil {
  90                 typeData := jm.allTypes[ji.InfoTypeIdentity]
  91                 typeData.jobHandler.addJobCh <- ji
  92                 log.Debug("Added job: ", ji)
  93                 return nil
  94         } else {
  95                 return err
  96         }
  97 }
  98
  99 func (jm *JobsManagerImpl) DeleteJob(jobId string) {
 100         for _, typeData := range jm.allTypes {
 101                 log.Debugf("Deleting job %v from type %v", jobId, typeData.TypeId)
 102                 typeData.jobHandler.deleteJobCh <- jobId
 103         }
 104         log.Debug("Deleted job: ", jobId)
 105 }
 106
 107 func (jm *JobsManagerImpl) validateJobInfo(ji JobInfo) error {
 108         if _, ok := jm.allTypes[ji.InfoTypeIdentity]; !ok {
 109                 return fmt.Errorf("type not supported: %v", ji.InfoTypeIdentity)
 110         }
 111         if ji.InfoJobIdentity == "" {
 112                 return fmt.Errorf("missing required job identity: %v", ji)
 113         }
 114         // Temporary for when there are only REST callbacks needed
 115         if ji.TargetUri == "" {
 116                 return fmt.Errorf("missing required target URI: %v", ji)
 117         }
 118         return nil
 119 }
 120
 121 func (jm *JobsManagerImpl) LoadTypesFromConfiguration() ([]config.TypeDefinition, error) {
 122         typeDefsByte, err := os.ReadFile(jm.configFile)
 123         if err != nil {
 124                 return nil, err
 125         }
 126         typeDefs := struct {
 127                 Types []config.TypeDefinition `json:"types"`
 128         }{}
 129         err = json.Unmarshal(typeDefsByte, &typeDefs)
 130         if err != nil {
 131                 return nil, err
 132         }
 133         for _, typeDef := range typeDefs.Types {
 134                 addCh := make(chan JobInfo)
 135                 deleteCh := make(chan string)
 136                 jh := jobHandler{
 137                         typeId:           typeDef.Id,
 138                         topicUrl:         typeDef.DmaapTopicURL,
 139                         jobs:             make(map[string]JobInfo),
 140                         addJobCh:         addCh,
 141                         deleteJobCh:      deleteCh,
 142                         pollClient:       jm.pollClient,
 143                         distributeClient: jm.distributeClient,
 144                 }
 145                 jm.allTypes[typeDef.Id] = TypeData{
 146                         TypeId:        typeDef.Id,
 147                         DMaaPTopicURL: typeDef.DmaapTopicURL,
 148                         jobHandler:    &jh,
 149                 }
 150         }
 151         return typeDefs.Types, nil
 152 }
 153
 154 func (jm *JobsManagerImpl) GetSupportedTypes() []string {
 155         supportedTypes := []string{}
 156         for k := range jm.allTypes {
 157                 supportedTypes = append(supportedTypes, k)
 158         }
 159         return supportedTypes
 160 }
 161
 162 func (jm *JobsManagerImpl) StartJobs() {
 163         for _, jobType := range jm.allTypes {
 164
 165                 go jobType.jobHandler.start(jm.mrAddress)
 166
 167         }
 168 }
 169
 170 func (jh *jobHandler) start(mRAddress string) {
 171         go func() {
 172                 for {
 173                         jh.pollAndDistributeMessages(mRAddress)
 174                 }
 175         }()
 176
 177         go func() {
 178                 for {
 179                         jh.monitorManagementChannels()
 180                 }
 181         }()
 182 }
 183
 184 func (jh *jobHandler) pollAndDistributeMessages(mRAddress string) {
 185         log.Debugf("Processing jobs for type: %v", jh.typeId)
 186         messagesBody, error := restclient.Get(mRAddress+jh.topicUrl, jh.pollClient)
 187         if error != nil {
 188                 log.Warnf("Error getting data from MR. Cause: %v", error)
 189         }
 190         log.Debugf("Received messages: %v", string(messagesBody))
 191         jh.distributeMessages(messagesBody)
 192 }
 193
 194 func (jh *jobHandler) distributeMessages(messages []byte) {
 195         if len(messages) > 2 {
 196                 jh.mu.Lock()
 197                 defer jh.mu.Unlock()
 198                 for _, jobInfo := range jh.jobs {
 199                         go jh.sendMessagesToConsumer(messages, jobInfo)
 200                 }
 201         }
 202 }
 203
 204 func (jh *jobHandler) sendMessagesToConsumer(messages []byte, jobInfo JobInfo) {
 205         log.Debugf("Processing job: %v", jobInfo.InfoJobIdentity)
 206         if postErr := restclient.Post(jobInfo.TargetUri, messages, jh.distributeClient); postErr != nil {
 207                 log.Warnf("Error posting data for job: %v. Cause: %v", jobInfo, postErr)
 208         }
 209         log.Debugf("Messages distributed to consumer: %v.", jobInfo.Owner)
 210 }
 211
 212 func (jh *jobHandler) monitorManagementChannels() {
 213         select {
 214         case addedJob := <-jh.addJobCh:
 215                 jh.mu.Lock()
 216                 log.Debugf("received %v from addJobCh\n", addedJob)
 217                 jh.jobs[addedJob.InfoJobIdentity] = addedJob
 218                 jh.mu.Unlock()
 219         case deletedJob := <-jh.deleteJobCh:
 220                 jh.mu.Lock()
 221                 log.Debugf("received %v from deleteJobCh\n", deletedJob)
 222                 delete(jh.jobs, deletedJob)
 223                 jh.mu.Unlock()
 224         }
 225 }