Add status check to the Go versions of the O-RO Front-Haul Recovery and O-DU Slice Assurance
usecases.
Issue-ID: NONRTRIC-671
Signed-off-by: elinuxhenrik <henrik.b.andersson@est.tech>
Change-Id: Ic7e481b3406b97e607d935707d2c9c4c25b503a3
!consumer/
dmaap
!dmaap/
+ics
+!ics/
Once the initial registration is done, the producer will constantly poll MR for all configured job types. When receiving messages for a type, it will distribute these messages to all jobs registered for the type. If no jobs for that type are registered, the messages will be discarded. If a consumer is unavailable for distribution, the messages will be discarded for that consumer until it is available again.
-The producer provides a REST API to control the log level. The available levels are the same as the ones used in the configuration above.
+The producer provides a REST API that fulfills the ICS Data producer API, see [Data producer (callbacks)](<https://docs.o-ran-sc.org/projects/o-ran-sc-nonrtric/en/latest/ics-api.html#tag/Data-producer-(callbacks)>). The health check method returns the registration status of the producer in ICS as JSON. It also provides a method to control the log level of the producer. The available log levels are the same as the ones used in the configuration above.
PUT https://mrproducer:8085/admin/log?level=<new level>
"oransc.org/nonrtric/dmaapmediatorproducer/internal/jobs"
)
-const StatusPath = "/status"
-const AddJobPath = "/jobs"
+const HealthCheckPath = "/health_check"
+const AddJobPath = "/info_job"
const jobIdToken = "infoJobId"
const deleteJobPath = AddJobPath + "/{" + jobIdToken + "}"
const logLevelToken = "level"
}
}
-func NewRouter(jm jobs.JobsManager) *mux.Router {
+func NewRouter(jm jobs.JobsManager, hcf func(http.ResponseWriter, *http.Request)) *mux.Router {
callbackHandler := NewProducerCallbackHandler(jm)
r := mux.NewRouter()
- r.HandleFunc(StatusPath, statusHandler).Methods(http.MethodGet).Name("status")
+ r.HandleFunc(HealthCheckPath, hcf).Methods(http.MethodGet).Name("health_check")
r.HandleFunc(AddJobPath, callbackHandler.addInfoJobHandler).Methods(http.MethodPost).Name("add")
r.HandleFunc(deleteJobPath, callbackHandler.deleteInfoJobHandler).Methods(http.MethodDelete).Name("delete")
r.HandleFunc(logAdminPath, callbackHandler.setLogLevel).Methods(http.MethodPut).Name("setLogLevel")
return r
}
-func statusHandler(w http.ResponseWriter, r *http.Request) {
- // Just respond OK to show the server is alive for now. Might be extended later.
-}
-
func (h *ProducerCallbackHandler) addInfoJobHandler(w http.ResponseWriter, r *http.Request) {
b, readErr := ioutil.ReadAll(r.Body)
if readErr != nil {
func TestNewRouter(t *testing.T) {
assertions := require.New(t)
- r := NewRouter(nil)
- statusRoute := r.Get("status")
+ r := NewRouter(nil, nil)
+ statusRoute := r.Get("health_check")
assertions.NotNil(statusRoute)
supportedMethods, err := statusRoute.GetMethods()
assertions.Equal([]string{http.MethodGet}, supportedMethods)
assertions.Nil(err)
path, _ := statusRoute.GetPathTemplate()
- assertions.Equal("/status", path)
+ assertions.Equal("/health_check", path)
addJobRoute := r.Get("add")
assertions.NotNil(addJobRoute)
assertions.Equal([]string{http.MethodPost}, supportedMethods)
assertions.Nil(err)
path, _ = addJobRoute.GetPathTemplate()
- assertions.Equal("/jobs", path)
+ assertions.Equal("/info_job", path)
deleteJobRoute := r.Get("delete")
assertions.NotNil(deleteJobRoute)
assertions.Equal([]string{http.MethodDelete}, supportedMethods)
assertions.Nil(err)
path, _ = deleteJobRoute.GetPathTemplate()
- assertions.Equal("/jobs/{infoJobId}", path)
+ assertions.Equal("/info_job/{infoJobId}", path)
notFoundHandler := r.NotFoundHandler
handler := http.HandlerFunc(notFoundHandler.ServeHTTP)
assertions.Equal("/admin/log", path)
}
-func TestStatusHandler(t *testing.T) {
- assertions := require.New(t)
-
- handler := http.HandlerFunc(statusHandler)
- responseRecorder := httptest.NewRecorder()
- r := newRequest(http.MethodGet, "/status", nil, t)
-
- handler.ServeHTTP(responseRecorder, r)
-
- assertions.Equal(http.StatusOK, responseRecorder.Code)
- assertions.Equal("", responseRecorder.Body.String())
-}
-
func TestAddInfoJobHandler(t *testing.T) {
assertions := require.New(t)
)
var configuration *config.Config
+var registered bool
func init() {
configuration = config.New()
retryClient := restclient.CreateRetryClient(cert)
jobsManager := jobs.NewJobsManagerImpl(retryClient, configuration.DMaaPMRAddress, restclient.CreateClientWithoutRetry(cert, 10*time.Second))
+ go startCallbackServer(jobsManager, callbackAddress)
+
if err := registerTypesAndProducer(jobsManager, configuration.InfoCoordinatorAddress, callbackAddress, retryClient); err != nil {
log.Fatalf("Stopping producer due to: %v", err)
}
+ registered = true
jobsManager.StartJobsForAllTypes()
log.Debug("Starting DMaaP Mediator Producer")
- go func() {
- log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort)
- r := server.NewRouter(jobsManager)
- if restclient.IsUrlSecure(callbackAddress) {
- log.Fatalf("Server stopped: %v", http.ListenAndServeTLS(fmt.Sprintf(":%v", configuration.InfoProducerPort), configuration.ProducerCertPath, configuration.ProducerKeyPath, r))
- } else {
- log.Fatalf("Server stopped: %v", http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r))
- }
- }()
keepProducerAlive()
}
}
producer := config.ProducerRegistrationInfo{
- InfoProducerSupervisionCallbackUrl: callbackAddress + server.StatusPath,
+ InfoProducerSupervisionCallbackUrl: callbackAddress + server.HealthCheckPath,
SupportedInfoTypes: jobTypesHandler.GetSupportedTypes(),
InfoJobCallbackUrl: callbackAddress + server.AddJobPath,
}
return nil
}
+func startCallbackServer(jobsManager jobs.JobsManager, callbackAddress string) {
+ log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort)
+ r := server.NewRouter(jobsManager, statusHandler)
+ if restclient.IsUrlSecure(callbackAddress) {
+ log.Fatalf("Server stopped: %v", http.ListenAndServeTLS(fmt.Sprintf(":%v", configuration.InfoProducerPort), configuration.ProducerCertPath, configuration.ProducerKeyPath, r))
+ } else {
+ log.Fatalf("Server stopped: %v", http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r))
+ }
+}
+
+func statusHandler(w http.ResponseWriter, r *http.Request) {
+ registeredStatus := "not registered"
+ if registered {
+ registeredStatus = "registered"
+ }
+ fmt.Fprintf(w, `{"status": "%v"}`, registeredStatus)
+}
+
func keepProducerAlive() {
forever := make(chan int)
<-forever
}
fmt.Println("Registering consumer: ", jobInfo)
body, _ := json.Marshal(jobInfo)
- putErr := restclient.Put(fmt.Sprintf("http://localhost:8083/data-consumer/v1/info-jobs/job%v", port), body, &httpClient)
+ putErr := restclient.Put(fmt.Sprintf("https://localhost:8083/data-consumer/v1/info-jobs/job%v", port), body, &httpClient)
if putErr != nil {
fmt.Println("Unable to register consumer: ", putErr)
}
--- /dev/null
+// -
+// ========================LICENSE_START=================================
+// O-RAN-SC
+// %%
+// Copyright (C) 2021: Nordix Foundation
+// %%
+// 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 (
+ "flag"
+ "fmt"
+ "net/http"
+
+ "github.com/gorilla/mux"
+)
+
+func main() {
+ port := flag.Int("port", 8434, "The port this stub will listen on")
+ flag.Parse()
+ fmt.Println("Starting ICS stub on port ", *port)
+
+ r := mux.NewRouter()
+ r.HandleFunc("/data-producer/v1/info-types/{typeId}", handleTypeRegistration).Methods(http.MethodPut, http.MethodPut)
+ r.HandleFunc("/data-producer/v1/info-producers/{producerId}", handleProducerRegistration).Methods(http.MethodPut, http.MethodPut)
+ fmt.Println(http.ListenAndServe(fmt.Sprintf(":%v", *port), r))
+}
+
+func handleTypeRegistration(w http.ResponseWriter, r *http.Request) {
+ vars := mux.Vars(r)
+ id, ok := vars["typeId"]
+ if ok {
+ fmt.Println("Registered type ", id)
+ }
+}
+
+func handleProducerRegistration(w http.ResponseWriter, r *http.Request) {
+ vars := mux.Vars(r)
+ id, ok := vars["producerId"]
+ if ok {
+ fmt.Println("Registered producer ", id)
+ }
+}
-# O-RAN-SC Non-RealTime RIC O-DU Closed Loop Usecase Slice Assurance
+# O-RAN-SC Non-RealTime RIC O-DU Closed Loop Usecase Slice Assurance
## Configuration
>- LOG_LEVEL Optional. The log level, which can be `Error`, `Warn`, `Info` or `Debug`. Defaults to `Info`.
>- POLLTIME Optional. Waiting time between one pull request to Dmaap and another. Defaults to 10 sec
+## Functionality
+
+There is a status call provided in a REST API on port 40936.
+>- /status OK
## Development
To make it easy to test during development of the consumer, there is a stub provided in the `stub` folder.
-This stub is used to simulate both received VES messages from Dmaap MR with information about performance measurements for the slices in a determinated DU and also SDNR, that sends information about Radio Resource Management Policy Ratio and allows to modify value for RRM Policy Dedicated Ratio from default to higher value.
+This stub is used to simulate both received VES messages from Dmaap MR with information about performance measurements for the slices in a determinated DU and also SDNR, that sends information about Radio Resource Management Policy Ratio and allows to modify value for RRM Policy Dedicated Ratio from default to higher value.
By default, SDNR stub listens to the port `3904`, but his can be overridden by passing a `--sdnr-port [PORT]` flag when starting the stub. For Dmaap MR stub default port is `3905` but it can be overriden by passing a `--dmaap-port [PORT]` flag when starting the stub.
}
func (c Config) String() string {
- return fmt.Sprintf("ConsumerHost: %v, ConsumerPort: %v, SDNRAddress: %v, SDNRUser: %v, SDNRPassword: %v, LogLevel: %v", c.MRHost, c.MRPort, c.SDNRAddress, c.SDNRUser, c.SDNPassword, c.LogLevel)
+ return fmt.Sprintf("[MRHost: %v, MRPort: %v, SDNRAddress: %v, SDNRUser: %v, SDNRPassword: %v, PollTime: %v, LogLevel: %v]", c.MRHost, c.MRPort, c.SDNRAddress, c.SDNRUser, c.SDNPassword, c.Polltime, c.LogLevel)
}
func getEnv(key string, defaultVal string) string {
import (
"fmt"
+ "net/http"
log "github.com/sirupsen/logrus"
"oransc.org/usecase/oduclosedloop/internal/config"
a := sliceassurance.App{}
a.Initialize(dmaapUrl, configuration.SDNRAddress)
- a.Run(TOPIC, configuration.Polltime)
+ go a.Run(TOPIC, configuration.Polltime)
+ http.HandleFunc("/status", statusHandler)
+
+ log.Fatal(http.ListenAndServe(":40936", nil))
}
func validateConfiguration(configuration *config.Config) error {
}
return nil
}
+
+func statusHandler(w http.ResponseWriter, r *http.Request) {
+ // Just respond OK to show the service is alive for now. Might be extended later.
+}
## Functionality
-The creation of the job is not done when the consumer is started. Instead the consumer provides a REST API where it can be started and stopped, described below.
+The creation of the job is not done when the consumer is started. Instead the consumer provides a REST API where it can be started and stopped, described below. The API is available on the host and port configured for the consumer
->- /start Creates the job in ICS.
->- /stop Deletes the job in ICS.
+>- /admin/start Creates the job in ICS.
+>- /admin/stop Deletes the job in ICS.
If the consumer is shut down with a SIGTERM, it will also delete the job before exiting.
+There is also a status call provided in the REST API. This will return the running status of the consumer as JSON.
+>- /status {"status": "started/stopped"}
+
## Development
To make it easy to test during development of the consumer, three stubs are provided in the `stub` folder.
}
func (c Config) String() string {
- return fmt.Sprintf("ConsumerHost: %v, ConsumerPort: %v, InfoCoordinatorAddress: %v, SDNRAddress: %v, SDNRUser: %v, SDNRPassword: %v, ORUToODUMapFile: %v, ConsumerCertPath: %v, ConsumerKeyPath: %v, LogLevel: %v", c.ConsumerHost, c.ConsumerPort, c.InfoCoordinatorAddress, c.SDNRAddress, c.SDNRUser, c.SDNPassword, c.ORUToODUMapFile, c.ConsumerCertPath, c.ConsumerKeyPath, c.LogLevel)
+ return fmt.Sprintf("{ConsumerHost: %v, ConsumerPort: %v, InfoCoordinatorAddress: %v, SDNRAddress: %v, SDNRUser: %v, SDNRPassword: %v, ORUToODUMapFile: %v, ConsumerCertPath: %v, ConsumerKeyPath: %v, LogLevel: %v}", c.ConsumerHost, c.ConsumerPort, c.InfoCoordinatorAddress, c.SDNRAddress, c.SDNRUser, c.SDNPassword, c.ORUToODUMapFile, c.ConsumerCertPath, c.ConsumerKeyPath, c.LogLevel)
}
func getEnv(key string, defaultVal string) string {
var linkfailureConfig linkfailure.Configuration
var lookupService repository.LookupService
var consumerPort string
+var started bool
func init() {
doInit()
r := mux.NewRouter()
r.HandleFunc("/", messageHandler.MessagesHandler).Methods(http.MethodPost).Name("messageHandler")
+ r.HandleFunc("/status", statusHandler).Methods(http.MethodGet).Name("status")
r.HandleFunc("/admin/start", startHandler).Methods(http.MethodPost).Name("start")
r.HandleFunc("/admin/stop", stopHandler).Methods(http.MethodPost).Name("stop")
return
}
log.Debug("Registered job.")
+ started = true
}
func stopHandler(w http.ResponseWriter, r *http.Request) {
return
}
log.Debug("Deleted job.")
+ started = false
+}
+
+func statusHandler(w http.ResponseWriter, r *http.Request) {
+ runStatus := "started"
+ if !started {
+ runStatus = "stopped"
+ }
+ fmt.Fprintf(w, `{"status": "%v"}`, runStatus)
}
func deleteOnShutdown(s chan os.Signal) {
assertions.Nil(err)
path, _ = stopHandlerRoute.GetPathTemplate()
assertions.Equal("/admin/stop", path)
+
+ statusHandlerRoute := r.Get("status")
+ assertions.NotNil(statusHandlerRoute)
+ supportedMethods, err = statusHandlerRoute.GetMethods()
+ assertions.Equal([]string{http.MethodGet}, supportedMethods)
+ assertions.Nil(err)
+ path, _ = statusHandlerRoute.GetPathTemplate()
+ assertions.Equal("/status", path)
}
func Test_startHandler(t *testing.T) {
expectedBody := wantedBody
assertions.Equal(expectedBody, body)
clientMock.AssertNumberOfCalls(t, "Do", 1)
+
+ // Check that the running status is "started"
+ statusHandler := http.HandlerFunc(statusHandler)
+ statusResponseRecorder := httptest.NewRecorder()
+ statusRequest, _ := http.NewRequest(http.MethodGet, "/status", nil)
+
+ statusHandler.ServeHTTP(statusResponseRecorder, statusRequest)
+
+ assertions.Equal(http.StatusOK, statusResponseRecorder.Code)
+ assertions.Equal(`{"status": "started"}`, statusResponseRecorder.Body.String())
})
}
}
assertions.Equal("enrichmentservice:8083", actualRequest.URL.Host)
assertions.Equal("/data-consumer/v1/info-jobs/14e7bb84-a44d-44c1-90b7-6995a92ad43c", actualRequest.URL.Path)
clientMock.AssertNumberOfCalls(t, "Do", 1)
+
+ // Check that the running status is "stopped"
+ statusHandler := http.HandlerFunc(statusHandler)
+ statusResponseRecorder := httptest.NewRecorder()
+ statusRequest, _ := http.NewRequest(http.MethodGet, "/status", nil)
+
+ statusHandler.ServeHTTP(statusResponseRecorder, statusRequest)
+
+ assertions.Equal(http.StatusOK, statusResponseRecorder.Code)
+ assertions.Equal(`{"status": "stopped"}`, statusResponseRecorder.Body.String())
})
}
}
func main() {
port := flag.Int("port", 8083, "The port this consumer will listen on")
flag.Parse()
- fmt.Println("Starting SDNR stub on port ", *port)
+ fmt.Println("Starting ICS stub on port ", *port)
r := mux.NewRouter()
r.HandleFunc("/data-consumer/v1/info-jobs/{jobId}", handleCalls).Methods(http.MethodPut, http.MethodDelete)
Code Review / nonrtric.git / commitdiff