pkg/xapp/xapp.go

   1 /*
   2 ==================================================================================
   3   Copyright (c) 2019 AT&T Intellectual Property.
   4   Copyright (c) 2019 Nokia
   5
   6    Licensed under the Apache License, Version 2.0 (the "License");
   7    you may not use this file except in compliance with the License.
   8    You may obtain a copy of the License at
   9
  10        http://www.apache.org/licenses/LICENSE-2.0
  11
  12    Unless required by applicable law or agreed to in writing, software
  13    distributed under the License is distributed on an "AS IS" BASIS,
  14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15    See the License for the specific language governing permissions and
  16    limitations under the License.
  17 ==================================================================================
  18 */
  19
  20 package xapp
  21
  22 import (
  23         "fmt"
  24         "github.com/spf13/viper"
  25         "net/http"
  26         "os"
  27         "os/signal"
  28         "sync/atomic"
  29         "syscall"
  30         "time"
  31 )
  32
  33 type ReadyCB func(interface{})
  34 type ShutdownCB func()
  35
  36 var (
  37         // XApp is an application instance
  38         Rmr           *RMRClient
  39         Sdl           *SDLClient
  40         Rnib          *RNIBClient
  41         Resource      *Router
  42         Metric        *Metrics
  43         Logger        *Log
  44         Config        Configurator
  45         Subscription  *Subscriber
  46         Alarm         *AlarmClient
  47         readyCb       ReadyCB
  48         readyCbParams interface{}
  49         shutdownCb    ShutdownCB
  50         shutdownFlag  int32
  51         shutdownCnt   int32
  52 )
  53
  54 func IsReady() bool {
  55         return Rmr != nil && Rmr.IsReady() && Sdl != nil && Sdl.IsReady()
  56 }
  57
  58 func SetReadyCB(cb ReadyCB, params interface{}) {
  59         readyCb = cb
  60         readyCbParams = params
  61 }
  62
  63 func xappReadyCb(params interface{}) {
  64         Alarm = NewAlarmClient(viper.GetString("alarm.MOId"), viper.GetString("alarm.APPId"))
  65         if readyCb != nil {
  66                 readyCb(readyCbParams)
  67         }
  68 }
  69
  70 func SetShutdownCB(cb ShutdownCB) {
  71         shutdownCb = cb
  72 }
  73
  74 func init() {
  75         // Load xapp configuration
  76         Logger = LoadConfig()
  77
  78         Logger.SetLevel(viper.GetInt("logger.level"))
  79         Resource = NewRouter()
  80         Config = Configurator{}
  81         Metric = NewMetrics(viper.GetString("metrics.url"), viper.GetString("metrics.namespace"), Resource.router)
  82         Subscription = NewSubscriber(viper.GetString("subscription.host"), viper.GetInt("subscription.timeout"))
  83
  84         if viper.IsSet("db.namespaces") {
  85                 namespaces := viper.GetStringSlice("db.namespaces")
  86                 if len(namespaces) > 0 && namespaces[0] != "" {
  87                         Sdl = NewSDLClient(viper.GetStringSlice("db.namespaces")[0])
  88                 }
  89                 if len(namespaces) > 1 && namespaces[1] != "" {
  90                         Rnib = NewRNIBClient(viper.GetStringSlice("db.namespaces")[1])
  91                 }
  92         } else {
  93                 Sdl = NewSDLClient(viper.GetString("db.namespace"))
  94         }
  95
  96         //
  97         // Signal handlers to really exit program.
  98         // shutdownCb can hang until application has
  99         // made all needed gracefull shutdown actions
 100         // hardcoded limit for shutdown is 20 seconds
 101         //
 102         interrupt := make(chan os.Signal, 1)
 103         signal.Notify(interrupt, syscall.SIGINT, syscall.SIGTERM)
 104         //signal handler function
 105         go func() {
 106                 for _ = range interrupt {
 107                         if atomic.CompareAndSwapInt32(&shutdownFlag, 0, 1) {
 108                                 // close function
 109                                 go func() {
 110                                         timeout := int(20)
 111                                         sentry := make(chan struct{})
 112                                         defer close(sentry)
 113
 114                                         // close callback
 115                                         go func() {
 116                                                 if shutdownCb != nil {
 117                                                         shutdownCb()
 118                                                 }
 119                                                 sentry <- struct{}{}
 120                                         }()
 121                                         select {
 122                                         case <-time.After(time.Duration(timeout) * time.Second):
 123                                                 Logger.Info("xapp-frame shutdown callback took more than %d seconds", timeout)
 124                                         case <-sentry:
 125                                                 Logger.Info("xapp-frame shutdown callback handled within %d seconds", timeout)
 126                                         }
 127                                         os.Exit(0)
 128                                 }()
 129                         } else {
 130                                 newCnt := atomic.AddInt32(&shutdownCnt, 1)
 131                                 Logger.Info("xapp-frame shutdown already ongoing. Forced exit counter %d/%d ", newCnt, 5)
 132                                 if newCnt >= 5 {
 133                                         Logger.Info("xapp-frame shutdown forced exit")
 134                                         os.Exit(0)
 135                                 }
 136                                 continue
 137                         }
 138
 139                 }
 140         }()
 141 }
 142
 143 func RunWithParams(c MessageConsumer, sdlcheck bool) {
 144         Rmr = NewRMRClient()
 145         Rmr.SetReadyCB(xappReadyCb, nil)
 146         go http.ListenAndServe(viper.GetString("local.host"), Resource.router)
 147         Logger.Info(fmt.Sprintf("Xapp started, listening on: %s", viper.GetString("local.host")))
 148         if sdlcheck {
 149                 Sdl.TestConnection()
 150         }
 151         Rmr.Start(c)
 152 }
 153
 154 func Run(c MessageConsumer) {
 155         RunWithParams(c, true)
 156 }