X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=RIC-E2-TERMINATION%2FsctpThread.cpp;h=ddcaae23b10f48ca06b1c426b5c8f4a054bdc449;hb=c18fe539422c5354bd5d3d0058a65bd1feecf5bf;hp=95b0e468afa782c76a79af4162b039f3e55b6278;hpb=7b437f758a65cbb98708e88ce72ccd1b2b0ea7fb;p=ric-plt%2Fe2.git diff --git a/RIC-E2-TERMINATION/sctpThread.cpp b/RIC-E2-TERMINATION/sctpThread.cpp index 95b0e46..ddcaae2 100644 --- a/RIC-E2-TERMINATION/sctpThread.cpp +++ b/RIC-E2-TERMINATION/sctpThread.cpp @@ -24,21 +24,15 @@ #include "sctpThread.h" #include "BuildRunName.h" -//#define asn_DEF_E2SM_gNB_NRT_RANfunction_Definition e2sm_asn_DEF_E2SM_gNB_NRT_RANfunction_Definition -//#define __gcov_init e2sm___gcov_init -//#define __gcov_merge_add e2sm___gcov_merge_add -//#define calloc e2sm_calloc - -#include "3rdparty/oranE2SM/E2SM-gNB-NRT-RANfunction-Definition.h" - -//#undef calloc -//#undef __gcov_merge_add -//#undef __gcov_init -//#undef asn_DEF_E2SM_gNB_NRT_RANfunction_Definition +//#include "3rdparty/oranE2SM/E2SM-gNB-NRT-RANfunction-Definition.h" +//#include "BuildXml.h" +//#include "pugixml/src/pugixml.hpp" using namespace std; //using namespace std::placeholders; using namespace boost::filesystem; +using namespace prometheus; + //#ifdef __cplusplus //extern "C" @@ -91,13 +85,18 @@ std::atomic num_of_XAPP_messages{0}; static long transactionCounter = 0; int buildListeningPort(sctp_params_t &sctpParams) { - sctpParams.listenFD = socket (AF_INET6, SOCK_STREAM, IPPROTO_SCTP); + sctpParams.listenFD = socket(AF_INET6, SOCK_STREAM, IPPROTO_SCTP); + if (sctpParams.listenFD <= 0) { + mdclog_write(MDCLOG_ERR, "Error Opening socket, %s", strerror(errno)); + return -1; + } + struct sockaddr_in6 servaddr {}; servaddr.sin6_family = AF_INET6; servaddr.sin6_addr = in6addr_any; servaddr.sin6_port = htons(sctpParams.sctpPort); if (bind(sctpParams.listenFD, (SA *)&servaddr, sizeof(servaddr)) < 0 ) { - mdclog_write(MDCLOG_ERR, "Error binding. %s\n", strerror(errno)); + mdclog_write(MDCLOG_ERR, "Error binding port %d. %s", sctpParams.sctpPort, strerror(errno)); return -1; } if (setSocketNoBlocking(sctpParams.listenFD) == -1) { @@ -245,6 +244,13 @@ int buildConfiguration(sctp_params_t &sctpParams) { } jsonTrace = sctpParams.trace; + sctpParams.epollTimeOut = -1; + + tmpStr = conf.getStringValue("prometheusPort"); + if (tmpStr.length() != 0) { + sctpParams.prometheusPort = tmpStr; + } + sctpParams.ka_message_length = snprintf(sctpParams.ka_message, KA_MESSAGE_SIZE, "{\"address\": \"%s:%d\"," "\"fqdn\": \"%s\"," "\"pod_name\": \"%s\"}", @@ -290,7 +296,20 @@ int buildConfiguration(sctp_params_t &sctpParams) { return 0; } +void startPrometheus(sctp_params_t &sctpParams) { + sctpParams.prometheusFamily = &BuildCounter() + .Name("E2T") + .Help("E2T message counter") + .Labels({{"POD_NAME", sctpParams.podName}}) + .Register(*sctpParams.prometheusRegistry); + string promethusPath = sctpParams.prometheusPort + "," + "[::]:" + sctpParams.prometheusPort; + if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "Start Prometheus Pull mode on %s", promethusPath.c_str()); + } + sctpParams.prometheusExposer = new Exposer(promethusPath, 1); + sctpParams.prometheusExposer->RegisterCollectable(sctpParams.prometheusRegistry); +} int main(const int argc, char **argv) { sctp_params_t sctpParams; @@ -333,6 +352,13 @@ int main(const int argc, char **argv) { exit(-1); } + //auto registry = std::make_shared(); + sctpParams.prometheusRegistry = std::make_shared(); + + //sctpParams.promtheusFamily = new Family("E2T", "E2T message counter", {{"E", sctpParams.podName}}); + + startPrometheus(sctpParams); + // start epoll sctpParams.epoll_fd = epoll_create1(0); if (sctpParams.epoll_fd == -1) { @@ -378,8 +404,6 @@ int main(const int argc, char **argv) { } } - auto statFlag = false; - auto statThread = std::thread(statColectorThread, (void *)&statFlag); //loop over term_init until first message from xApp handleTermInit(sctpParams); @@ -388,9 +412,6 @@ int main(const int argc, char **argv) { t.join(); } - statFlag = true; - statThread.join(); - return 0; } @@ -429,7 +450,7 @@ void sendTermInit(sctp_params_t &sctpParams) { rmr_bytes2xact(msg, tx, txLen); msg = rmr_send_msg(sctpParams.rmrCtx, msg); if (msg == nullptr) { - msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.myIP.length()); + msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.ka_message_length); } else if (msg->state == 0) { rmr_free_msg(msg); if (mdclog_level_get() >= MDCLOG_INFO) { @@ -538,7 +559,6 @@ void listener(sctp_params_t *params) { mdclog_write(MDCLOG_DEBUG, "started thread number %s", tid); } - RmrMessagesBuffer_t rmrMessageBuffer{}; //create and init RMR rmrMessageBuffer.rmrCtx = params->rmrCtx; @@ -565,20 +585,23 @@ void listener(sctp_params_t *params) { // rmrMessageBuffer.sendBufferedMessages[i] = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE); // } - message.statCollector = StatCollector::GetInstance(); - while (true) { if (mdclog_level_get() >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Start EPOLL Wait"); + mdclog_write(MDCLOG_DEBUG, "Start EPOLL Wait. Timeout = %d", params->epollTimeOut); } - auto numOfEvents = epoll_wait(params->epoll_fd, events, MAXEVENTS, -1); - if (numOfEvents < 0 && errno == EINTR) { + auto numOfEvents = epoll_wait(params->epoll_fd, events, MAXEVENTS, params->epollTimeOut); + if (numOfEvents == 0) { // time out if (mdclog_level_get() >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "got EINTR : %s", strerror(errno)); + mdclog_write(MDCLOG_DEBUG, "got epoll timeout"); } continue; - } - if (numOfEvents < 0) { + } else if (numOfEvents < 0) { + if (errno == EINTR) { + if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "got EINTR : %s", strerror(errno)); + } + continue; + } mdclog_write(MDCLOG_ERR, "Epoll wait failed, errno = %s", strerror(errno)); return; } @@ -726,12 +749,16 @@ void handleConfigChange(sctp_params_t *sctpParams) { // not the directory } if (event->len) { - if (!(sctpParams->configFileName.compare(event->name))) { + auto retVal = strcmp(sctpParams->configFileName.c_str(), event->name); + if (retVal != 0) { continue; } } // only the file we want if (event->mask & (uint32_t)IN_CLOSE_WRITE) { + if (mdclog_level_get() >= MDCLOG_INFO) { + mdclog_write(MDCLOG_INFO, "Configuration file changed"); + } if (exists(p)) { const int size = 2048; auto fileSize = file_size(p); @@ -795,6 +822,8 @@ void handleConfigChange(sctp_params_t *sctpParams) { sctpParams->trace = false; } jsonTrace = sctpParams->trace; + + endlessLoop = false; } } @@ -987,7 +1016,6 @@ int sendSctpMsg(ConnectedCU_t *peerInfo, ReportingMessages_t &message, Sctp_Map_ m->erase(key); return -1; } - message.statCollector->incSentMessage(string(message.message.enodbName)); message.message.direction = 'D'; // send report.buffer of size buildJsonMessage(message); @@ -1008,8 +1036,6 @@ int sendSctpMsg(ConnectedCU_t *peerInfo, ReportingMessages_t &message, Sctp_Map_ * @param rmrMessageBuffer */ void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) { - rmr_get_meid(rmrMessageBuffer.rcvMessage, (unsigned char *) (message.message.enodbName)); - message.message.asndata = rmrMessageBuffer.rcvMessage->payload; message.message.asnLength = rmrMessageBuffer.rcvMessage->len; @@ -1045,14 +1071,12 @@ int receiveDataFromSctp(struct epoll_event *events, // get the identity of the interface message.peerInfo = (ConnectedCU_t *)events->data.ptr; - message.statCollector = StatCollector::GetInstance(); struct timespec start{0, 0}; struct timespec decodestart{0, 0}; struct timespec end{0, 0}; E2AP_PDU_t *pdu = nullptr; - while (true) { if (loglevel >= MDCLOG_DEBUG) { mdclog_write(MDCLOG_DEBUG, "Start Read from SCTP %d fd", message.peerInfo->fileDescriptor); @@ -1069,7 +1093,6 @@ int receiveDataFromSctp(struct epoll_event *events, } memcpy(message.message.enodbName, message.peerInfo->enodbName, sizeof(message.peerInfo->enodbName)); - message.statCollector->incRecvMessage(string(message.message.enodbName)); message.message.direction = 'U'; message.message.time.tv_nsec = ts.tv_nsec; message.message.time.tv_sec = ts.tv_sec; @@ -1118,7 +1141,6 @@ int receiveDataFromSctp(struct epoll_event *events, if (rval.code != RC_OK) { mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) E2AP PDU from RAN : %s", rval.code, message.peerInfo->enodbName); - //todo may need reset to pdu break; } @@ -1158,22 +1180,12 @@ int receiveDataFromSctp(struct epoll_event *events, message.peerInfo->enodbName, end.tv_sec - decodestart.tv_sec, end.tv_nsec - decodestart.tv_nsec); } numOfMessages++; - // remove the break for EAGAIN - //break; if (pdu != nullptr) { - //TODO need to test ASN_STRUCT_RESET(asn_DEF_E2AP_PDU, pdu); to get better performance ASN_STRUCT_RESET(asn_DEF_E2AP_PDU, pdu); //ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu); //pdu = nullptr; } - //clock_gettime(CLOCK_MONOTONIC, &start); } - // in case of break to avoid memory leak -// if (pdu != nullptr) { -// //ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu); -// ASN_STRUCT_RESET(asn_DEF_E2AP_PDU, pdu); -// //pdu = nullptr; -// } if (done) { if (loglevel >= MDCLOG_INFO) { @@ -1207,78 +1219,354 @@ int receiveDataFromSctp(struct epoll_event *events, static void buildAndsendSetupRequest(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer, - E2AP_PDU_t *pdu) { + E2AP_PDU_t *pdu/*, + string const &messageName, + string const &ieName, + vector &functionsToAdd_v, + vector &functionsToModified_v*/) { auto logLevel = mdclog_level_get(); - // now we can send the data to e2Mgr - auto buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2; - auto *rmrMsg = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, buffer_size); - // add addrees to message - - - // unsigned char *buffer = &rmrMsg->payload[j]; + asn_enc_rval_t er; + auto buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2; unsigned char buffer[RECEIVE_SCTP_BUFFER_SIZE * 2]; + while (true) { + er = asn_encode_to_buffer(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, pdu, buffer, buffer_size); + if (er.encoded == -1) { + mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno)); + return; + } else if (er.encoded > (ssize_t) buffer_size) { + buffer_size = er.encoded + 128; + mdclog_write(MDCLOG_WARN, "Buffer of size %d is to small for %s. Reallocate buffer of size %d", + (int) buffer_size, + asn_DEF_E2AP_PDU.name, buffer_size); + buffer_size = er.encoded + 128; +// free(buffer); + continue; + } + buffer[er.encoded] = '\0'; + break; + } // encode to xml - auto er = asn_encode_to_buffer(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, pdu, buffer, buffer_size); - if (er.encoded == -1) { - mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno)); - } else if (er.encoded > (ssize_t) buffer_size) { - mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s", - (int) buffer_size, - asn_DEF_E2AP_PDU.name); - } else { - rmrMsg->len = snprintf((char *)rmrMsg->payload, RECEIVE_SCTP_BUFFER_SIZE * 2, "%s:%d|%s", + + string res((char *)buffer); + res.erase(std::remove(res.begin(), res.end(), '\n'), res.end()); + res.erase(std::remove(res.begin(), res.end(), '\t'), res.end()); + res.erase(std::remove(res.begin(), res.end(), ' '), res.end()); + +// string res {}; +// if (!functionsToAdd_v.empty() || !functionsToModified_v.empty()) { +// res = buildXmlData(messageName, ieName, functionsToAdd_v, functionsToModified_v, buffer, (size_t) er.encoded); +// } + rmr_mbuf_t *rmrMsg; +// if (res.length() == 0) { +// rmrMsg = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, buffer_size + 256); +// rmrMsg->len = snprintf((char *) rmrMsg->payload, RECEIVE_SCTP_BUFFER_SIZE * 2, "%s:%d|%s", +// message.peerInfo->sctpParams->myIP.c_str(), +// message.peerInfo->sctpParams->rmrPort, +// buffer); +// } else { + rmrMsg = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, (int)res.length() + 256); + rmrMsg->len = snprintf((char *) rmrMsg->payload, res.length() + 256, "%s:%d|%s", message.peerInfo->sctpParams->myIP.c_str(), message.peerInfo->sctpParams->rmrPort, - buffer); - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Setup request of size %d :\n %s\n", rmrMsg->len, rmrMsg->payload); - } - // send to RMR - message.message.messageType = rmrMsg->mtype = RIC_E2_SETUP_REQ; - rmrMsg->state = 0; - rmr_bytes2meid(rmrMsg, (unsigned char *) message.message.enodbName, strlen(message.message.enodbName)); + res.c_str()); +// } - static unsigned char tx[32]; - snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++); - rmr_bytes2xact(rmrMsg, tx, strlen((const char *) tx)); - - rmrMsg = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMsg); - if (rmrMsg == nullptr) { - mdclog_write(MDCLOG_ERR, "RMR failed to send returned nullptr"); - } else if (rmrMsg->state != 0) { - char meid[RMR_MAX_MEID]{}; - if (rmrMsg->state == RMR_ERR_RETRY) { - usleep(5); - rmrMsg->state = 0; - mdclog_write(MDCLOG_INFO, "RETRY sending Message %d to Xapp from %s", - rmrMsg->mtype, rmr_get_meid(rmrMsg, (unsigned char *) meid)); - rmrMsg = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMsg); - if (rmrMsg == nullptr) { - mdclog_write(MDCLOG_ERR, "RMR failed send returned nullptr"); - } else if (rmrMsg->state != 0) { - mdclog_write(MDCLOG_ERR, - "RMR Retry failed %s sending request %d to Xapp from %s", - translateRmrErrorMessages(rmrMsg->state).c_str(), - rmrMsg->mtype, - rmr_get_meid(rmrMsg, (unsigned char *) meid)); - } - } else { - mdclog_write(MDCLOG_ERR, "RMR failed: %s. sending request %d to Xapp from %s", + if (logLevel >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "Setup request of size %d :\n %s\n", rmrMsg->len, rmrMsg->payload); + } + // send to RMR + rmrMsg->mtype = message.message.messageType; + rmrMsg->state = 0; + rmr_bytes2meid(rmrMsg, (unsigned char *) message.message.enodbName, strlen(message.message.enodbName)); + + static unsigned char tx[32]; + snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++); + rmr_bytes2xact(rmrMsg, tx, strlen((const char *) tx)); + + rmrMsg = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMsg); + if (rmrMsg == nullptr) { + mdclog_write(MDCLOG_ERR, "RMR failed to send returned nullptr"); + } else if (rmrMsg->state != 0) { + char meid[RMR_MAX_MEID]{}; + if (rmrMsg->state == RMR_ERR_RETRY) { + usleep(5); + rmrMsg->state = 0; + mdclog_write(MDCLOG_INFO, "RETRY sending Message %d to Xapp from %s", + rmrMsg->mtype, rmr_get_meid(rmrMsg, (unsigned char *) meid)); + rmrMsg = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMsg); + if (rmrMsg == nullptr) { + mdclog_write(MDCLOG_ERR, "RMR failed send returned nullptr"); + } else if (rmrMsg->state != 0) { + mdclog_write(MDCLOG_ERR, + "RMR Retry failed %s sending request %d to Xapp from %s", translateRmrErrorMessages(rmrMsg->state).c_str(), rmrMsg->mtype, rmr_get_meid(rmrMsg, (unsigned char *) meid)); } + } else { + mdclog_write(MDCLOG_ERR, "RMR failed: %s. sending request %d to Xapp from %s", + translateRmrErrorMessages(rmrMsg->state).c_str(), + rmrMsg->mtype, + rmr_get_meid(rmrMsg, (unsigned char *) meid)); } - message.peerInfo->gotSetup = true; - buildJsonMessage(message); - if (rmrMsg != nullptr) { - rmr_free_msg(rmrMsg); + } + message.peerInfo->gotSetup = true; + buildJsonMessage(message); + if (rmrMsg != nullptr) { + rmr_free_msg(rmrMsg); + } +} + +#if 0 +int RAN_Function_list_To_Vector(RANfunctions_List_t& list, vector &runFunXML_v) { + auto index = 0; + runFunXML_v.clear(); + for (auto j = 0; j < list.list.count; j++) { + auto *raNfunctionItemIEs = (RANfunction_ItemIEs_t *)list.list.array[j]; + if (raNfunctionItemIEs->id == ProtocolIE_ID_id_RANfunction_Item && + (raNfunctionItemIEs->value.present == RANfunction_ItemIEs__value_PR_RANfunction_Item)) { + // encode to xml + E2SM_gNB_NRT_RANfunction_Definition_t *ranFunDef = nullptr; + auto rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER, + &asn_DEF_E2SM_gNB_NRT_RANfunction_Definition, + (void **)&ranFunDef, + raNfunctionItemIEs->value.choice.RANfunction_Item.ranFunctionDefinition.buf, + raNfunctionItemIEs->value.choice.RANfunction_Item.ranFunctionDefinition.size); + if (rval.code != RC_OK) { + mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) E2SM message from : %s", + rval.code, + asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name); + return -1; + } + + auto xml_buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2; + unsigned char xml_buffer[RECEIVE_SCTP_BUFFER_SIZE * 2]; + memset(xml_buffer, 0, RECEIVE_SCTP_BUFFER_SIZE * 2); + // encode to xml + auto er = asn_encode_to_buffer(nullptr, + ATS_BASIC_XER, + &asn_DEF_E2SM_gNB_NRT_RANfunction_Definition, + ranFunDef, + xml_buffer, + xml_buffer_size); + if (er.encoded == -1) { + mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", + asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name, + strerror(errno)); + } else if (er.encoded > (ssize_t)xml_buffer_size) { + mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s, at %s line %d", + (int) xml_buffer_size, + asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name, __func__, __LINE__); + } else { + if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "Encoding E2SM %s PDU number %d : %s", + asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name, + index++, + xml_buffer); + } + + string runFuncs = (char *)(xml_buffer); + runFunXML_v.emplace_back(runFuncs); + } } } + return 0; +} +int collectServiceUpdate_RequestData(E2AP_PDU_t *pdu, + Sctp_Map_t *sctpMap, + ReportingMessages_t &message, + vector &RANfunctionsAdded_v, + vector &RANfunctionsModified_v) { + memset(message.peerInfo->enodbName, 0 , MAX_ENODB_NAME_SIZE); + for (auto i = 0; i < pdu->choice.initiatingMessage->value.choice.RICserviceUpdate.protocolIEs.list.count; i++) { + auto *ie = pdu->choice.initiatingMessage->value.choice.RICserviceUpdate.protocolIEs.list.array[i]; + if (ie->id == ProtocolIE_ID_id_RANfunctionsAdded) { + if (ie->value.present == RICserviceUpdate_IEs__value_PR_RANfunctionsID_List) { + if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "Run function list have %d entries", + ie->value.choice.RANfunctions_List.list.count); + } + if (RAN_Function_list_To_Vector(ie->value.choice.RANfunctions_List, RANfunctionsAdded_v) != 0 ) { + return -1; + } + } + } else if (ie->id == ProtocolIE_ID_id_RANfunctionsModified) { + if (ie->value.present == RICserviceUpdate_IEs__value_PR_RANfunctions_List) { + if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "Run function list have %d entries", + ie->value.choice.RANfunctions_List.list.count); + } + if (RAN_Function_list_To_Vector(ie->value.choice.RANfunctions_List, RANfunctionsModified_v) != 0 ) { + return -1; + } + } + } + } + if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "Run function vector have %ld entries", + RANfunctionsAdded_v.size()); + } + return 0; } + +#endif + + +void buildPrometheuslist(ConnectedCU_t *peerInfo, Family *prometheusFamily) { + peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"SetupRequest", "Messages"}}); + peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"SetupRequest", "Bytes"}}); + + peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ErrorIndication", "Messages"}}); + peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ErrorIndication", "Bytes"}}); + + peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_RICindication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICindication", "Messages"}}); + peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_RICindication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICindication", "Bytes"}}); + + peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetRequest", "Messages"}}); + peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetRequest", "Bytes"}}); + + peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICserviceUpdate", "Messages"}}); + peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICserviceUpdate", "Bytes"}}); + // --------------------------------------------- + peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetACK", "Messages"}}); + peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetACK", "Bytes"}}); + + peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolACK", "Messages"}}); + peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolACK", "Bytes"}}); + + peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionACK", "Messages"}}); + peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionACK", "Bytes"}}); + + peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteACK", "Messages"}}); + peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteACK", "Bytes"}}); + //------------------------------------------------------------- + + peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolFailure", "Messages"}}); + peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolFailure", "Bytes"}}); + + peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionFailure", "Messages"}}); + peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionFailure", "Bytes"}}); + + peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteFailure", "Messages"}}); + peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteFailure", "Bytes"}}); + + //==================================================================================== + peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ErrorIndication", "Messages"}}); + peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ErrorIndication", "Bytes"}}); + + peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetRequest", "Messages"}}); + peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetRequest", "Bytes"}}); + + peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICcontrol", "Messages"}}); + peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICcontrol", "Bytes"}}); + + peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICserviceQuery)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceQuery", "Messages"}}); + peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICserviceQuery)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceQuery", "Bytes"}}); + + peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscription", "Messages"}}); + peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscription", "Bytes"}}); + + peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscriptionDelete", "Messages"}}); + peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscriptionDelete", "Bytes"}}); + //--------------------------------------------------------------------------------------------------------- + peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupResponse", "Messages"}}); + peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupResponse", "Bytes"}}); + + peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetACK", "Messages"}}); + peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetACK", "Bytes"}}); + + peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateResponse", "Messages"}}); + peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateResponse", "Bytes"}}); + //---------------------------------------------------------------------------------------------------------------- + peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupRequestFailure", "Messages"}}); + peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupRequestFailure", "Bytes"}}); + + peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateFailure", "Messages"}}); + peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateFailure", "Bytes"}}); +} +/** + * + * @param pdu + * @param sctpMap + * @param message + * @param RANfunctionsAdded_v + * @return + */ +int collectSetupRequestData(E2AP_PDU_t *pdu, + Sctp_Map_t *sctpMap, + ReportingMessages_t &message /*, vector &RANfunctionsAdded_v*/) { + memset(message.peerInfo->enodbName, 0 , MAX_ENODB_NAME_SIZE); + for (auto i = 0; i < pdu->choice.initiatingMessage->value.choice.E2setupRequest.protocolIEs.list.count; i++) { + auto *ie = pdu->choice.initiatingMessage->value.choice.E2setupRequest.protocolIEs.list.array[i]; + if (ie->id == ProtocolIE_ID_id_GlobalE2node_ID) { + // get the ran name for meid + if (ie->value.present == E2setupRequestIEs__value_PR_GlobalE2node_ID) { + if (buildRanName(message.peerInfo->enodbName, ie) < 0) { + mdclog_write(MDCLOG_ERR, "Bad param in E2setupRequestIEs GlobalE2node_ID.\n"); + // no mesage will be sent + return -1; + } + + memcpy(message.message.enodbName, message.peerInfo->enodbName, strlen(message.peerInfo->enodbName)); + sctpMap->setkey(message.message.enodbName, message.peerInfo); + } + } /*else if (ie->id == ProtocolIE_ID_id_RANfunctionsAdded) { + if (ie->value.present == E2setupRequestIEs__value_PR_RANfunctions_List) { + if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "Run function list have %d entries", + ie->value.choice.RANfunctions_List.list.count); + } + if (RAN_Function_list_To_Vector(ie->value.choice.RANfunctions_List, RANfunctionsAdded_v) != 0 ) { + return -1; + } + } + } */ + } +// if (mdclog_level_get() >= MDCLOG_DEBUG) { +// mdclog_write(MDCLOG_DEBUG, "Run function vector have %ld entries", +// RANfunctionsAdded_v.size()); +// } + return 0; +} + +int XML_From_PER(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) { + E2AP_PDU_t *pdu = nullptr; + + if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "got PER message of size %d is:%s", + rmrMessageBuffer.sendMessage->len, rmrMessageBuffer.sendMessage->payload); + } + auto rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2AP_PDU, (void **) &pdu, + rmrMessageBuffer.sendMessage->payload, rmrMessageBuffer.sendMessage->len); + if (rval.code != RC_OK) { + mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) setup response from E2MGR : %s", + rval.code, + message.message.enodbName); + return -1; + } + + int buff_size = RECEIVE_XAPP_BUFFER_SIZE; + auto er = asn_encode_to_buffer(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, pdu, + rmrMessageBuffer.sendMessage->payload, buff_size); + if (er.encoded == -1) { + mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno)); + return -1; + } else if (er.encoded > (ssize_t)buff_size) { + mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s, at %s line %d", + (int)rmrMessageBuffer.sendMessage->len, + asn_DEF_E2AP_PDU.name, + __func__, + __LINE__); + return -1; + } + rmrMessageBuffer.sendMessage->len = er.encoded; + return 0; + +} + /** * * @param pdu @@ -1297,105 +1585,46 @@ void asnInitiatingRequest(E2AP_PDU_t *pdu, switch (procedureCode) { case ProcedureCode_id_E2setup: { if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got E2setup\n"); + mdclog_write(MDCLOG_DEBUG, "Got E2setup"); } - // first get the message as XML buffer - auto setup_xml_buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2; - unsigned char setup_xml_buffer[RECEIVE_SCTP_BUFFER_SIZE * 2]; - //unsigned char *tmp_buff_cursor = setup_xml_buffer; - - auto er = asn_encode_to_buffer(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, pdu, setup_xml_buffer, setup_xml_buffer_size); - if (er.encoded == -1) { - mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno)); - } else if (er.encoded > (ssize_t) setup_xml_buffer_size) { - mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s", - (int)setup_xml_buffer_size, - asn_DEF_E2AP_PDU.name); - } - std::string xmlString(setup_xml_buffer_size, setup_xml_buffer_size + er.encoded); - - auto failed = false; - memset(message.peerInfo->enodbName, 0 , MAX_ENODB_NAME_SIZE); - for (auto i = 0; i < pdu->choice.initiatingMessage->value.choice.E2setupRequest.protocolIEs.list.count; i++) { - auto *ie = pdu->choice.initiatingMessage->value.choice.E2setupRequest.protocolIEs.list.array[i]; - if (ie->id == ProtocolIE_ID_id_GlobalE2node_ID) { - // get the ran name for meid - if (ie->value.present == E2setupRequestIEs__value_PR_GlobalE2node_ID) { - if (buildRanName(message.peerInfo->enodbName, ie) < 0) { - mdclog_write(MDCLOG_ERR, "Bad param in E2setupRequestIEs GlobalE2node_ID.\n"); - // no mesage will be sent - break; - } - memcpy(message.message.enodbName, message.peerInfo->enodbName, strlen(message.peerInfo->enodbName)); - sctpMap->setkey(message.message.enodbName, message.peerInfo); - } - } - // reformat RANFUNCTION Definition to XML - if (ie->id == ProtocolIE_ID_id_RANfunctionsAdded) { - if (ie->value.present == E2setupRequestIEs__value_PR_RANfunctions_List) { - for (auto j = 0; i < ie->value.choice.RANfunctions_List.list.count; i++) { - auto *raNfunctionItemIEs = (RANfunction_ItemIEs_t *)ie->value.choice.RANfunctions_List.list.array[j]; - if (raNfunctionItemIEs->id == ProtocolIE_ID_id_RANfunction_Item) { - auto buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2; - unsigned char buffer[RECEIVE_SCTP_BUFFER_SIZE * 2]; - // encode to xml - E2SM_gNB_NRT_RANfunction_Definition_t *ranFunDef = nullptr; - auto rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER, - &asn_DEF_E2SM_gNB_NRT_RANfunction_Definition, - (void **)&ranFunDef, - buffer, - buffer_size); - if (rval.code != RC_OK) { - mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) E2AP PDU from E2MGR : %s", - rval.code, - asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name); - failed = true; - break; - } - - if (mdclog_level_get() >= MDCLOG_DEBUG) { - char *printBuffer; - size_t size; - FILE *stream = open_memstream(&printBuffer, &size); - asn_fprint(stream, &asn_DEF_E2AP_PDU, pdu); - mdclog_write(MDCLOG_DEBUG, "Encoding E2AP PDU past : %s", printBuffer); - } - auto xml_buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2; - unsigned char xml_buffer[RECEIVE_SCTP_BUFFER_SIZE * 2]; - // encode to xml - er = asn_encode_to_buffer(nullptr, - ATS_BASIC_XER, - &asn_DEF_E2SM_gNB_NRT_RANfunction_Definition, - ranFunDef, - xml_buffer, - xml_buffer_size); - if (er.encoded == -1) { - mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", - asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name, - strerror(errno)); - } else if (er.encoded > (ssize_t) buffer_size) { - mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s", - (int) xml_buffer_size, - asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name); - } else { - // we have the xml - } - - } - } - if (failed) { - break; - } - } - } - } - if (failed) { +// vector RANfunctionsAdded_v; +// vector RANfunctionsModified_v; +// RANfunctionsAdded_v.clear(); +// RANfunctionsModified_v.clear(); + if (collectSetupRequestData(pdu, sctpMap, message) != 0) { break; } - //build all parts and send the XML (need to copy the XML with the header to the rmrMessageBuffer payload - //TODO replace with new function + buildPrometheuslist(message.peerInfo, message.peerInfo->sctpParams->prometheusFamily); + + string messageName("E2setupRequest"); + string ieName("E2setupRequestIEs"); + message.message.messageType = RIC_E2_SETUP_REQ; + message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment(); + message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment((double)message.message.asnLength); + buildAndsendSetupRequest(message, rmrMessageBuffer, pdu); + break; + } + case ProcedureCode_id_RICserviceUpdate: { + if (logLevel >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "Got RICserviceUpdate %s", message.message.enodbName); + } +// vector RANfunctionsAdded_v; +// vector RANfunctionsModified_v; +// RANfunctionsAdded_v.clear(); +// RANfunctionsModified_v.clear(); +// if (collectServiceUpdate_RequestData(pdu, sctpMap, message, +// RANfunctionsAdded_v, RANfunctionsModified_v) != 0) { +// break; +// } + + string messageName("RICserviceUpdate"); + string ieName("RICserviceUpdateIEs"); + message.message.messageType = RIC_SERVICE_UPDATE; + message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(); + message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment((double)message.message.asnLength); + buildAndsendSetupRequest(message, rmrMessageBuffer, pdu); break; } @@ -1403,6 +1632,8 @@ void asnInitiatingRequest(E2AP_PDU_t *pdu, if (logLevel >= MDCLOG_DEBUG) { mdclog_write(MDCLOG_DEBUG, "Got ErrorIndication %s", message.message.enodbName); } + message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment(); + message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment((double)message.message.asnLength); if (sendRequestToXapp(message, RIC_ERROR_INDICATION, rmrMessageBuffer) != 0) { mdclog_write(MDCLOG_ERR, "RIC_ERROR_INDICATION failed to send to xAPP"); } @@ -1412,14 +1643,15 @@ void asnInitiatingRequest(E2AP_PDU_t *pdu, if (logLevel >= MDCLOG_DEBUG) { mdclog_write(MDCLOG_DEBUG, "Got Reset %s", message.message.enodbName); } - if (sendRequestToXapp(message, RIC_X2_RESET, rmrMessageBuffer) != 0) { - mdclog_write(MDCLOG_ERR, "RIC_X2_RESET message failed to send to xAPP"); + + message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment(); + message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength); + if (XML_From_PER(message, rmrMessageBuffer) < 0) { + break; } - break; - } - case ProcedureCode_id_RICcontrol: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICcontrol %s", message.message.enodbName); + + if (sendRequestToXapp(message, RIC_E2_RESET_REQ, rmrMessageBuffer) != 0) { + mdclog_write(MDCLOG_ERR, "RIC_E2_RESET_REQ message failed to send to xAPP"); } break; } @@ -1446,12 +1678,18 @@ void asnInitiatingRequest(E2AP_PDU_t *pdu, (unsigned char *)message.message.enodbName, strlen(message.message.enodbName)); rmrMessageBuffer.sendMessage->state = 0; - rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID; + rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricInstanceID; + + //ie->value.choice.RICrequestID.ricInstanceID; if (mdclog_level_get() >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "RIC sub id = %d, message type = %d", + mdclog_write(MDCLOG_DEBUG, "sub id = %d, mtype = %d, ric instance id %ld, requestor id = %ld", rmrMessageBuffer.sendMessage->sub_id, - rmrMessageBuffer.sendMessage->mtype); + rmrMessageBuffer.sendMessage->mtype, + ie->value.choice.RICrequestID.ricInstanceID, + ie->value.choice.RICrequestID.ricRequestorID); } + message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICindication]->Increment(); + message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICindication]->Increment((double)message.message.asnLength); sendRmrMessage(rmrMessageBuffer, message); messageSent = true; } else { @@ -1464,33 +1702,6 @@ void asnInitiatingRequest(E2AP_PDU_t *pdu, } break; } - case ProcedureCode_id_RICserviceQuery: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICserviceQuery %s", message.message.enodbName); - } - break; - } - case ProcedureCode_id_RICserviceUpdate: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICserviceUpdate %s", message.message.enodbName); - } - if (sendRequestToXapp(message, RIC_SERVICE_UPDATE, rmrMessageBuffer) != 0) { - mdclog_write(MDCLOG_ERR, "RIC_SERVICE_UPDATE message failed to send to xAPP"); - } - break; - } - case ProcedureCode_id_RICsubscription: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICsubscription %s", message.message.enodbName); - } - break; - } - case ProcedureCode_id_RICsubscriptionDelete: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICsubscriptionDelete %s", message.message.enodbName); - } - break; - } default: { mdclog_write(MDCLOG_ERR, "Undefined or not supported message = %ld", procedureCode); message.message.messageType = 0; // no RMR message type yet @@ -1518,27 +1729,17 @@ void asnSuccsesfulMsg(E2AP_PDU_t *pdu, mdclog_write(MDCLOG_INFO, "Successful Outcome %ld", procedureCode); } switch (procedureCode) { - case ProcedureCode_id_E2setup: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got E2setup\n"); - } - break; - } - case ProcedureCode_id_ErrorIndication: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got ErrorIndication %s", message.message.enodbName); - } - if (sendRequestToXapp(message, RIC_ERROR_INDICATION, rmrMessageBuffer) != 0) { - mdclog_write(MDCLOG_ERR, "RIC_ERROR_INDICATION failed to send to xAPP"); - } - break; - } case ProcedureCode_id_Reset: { if (logLevel >= MDCLOG_DEBUG) { mdclog_write(MDCLOG_DEBUG, "Got Reset %s", message.message.enodbName); } - if (sendRequestToXapp(message, RIC_X2_RESET, rmrMessageBuffer) != 0) { - mdclog_write(MDCLOG_ERR, "RIC_X2_RESET message failed to send to xAPP"); + message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment(); + message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength); + if (XML_From_PER(message, rmrMessageBuffer) < 0) { + break; + } + if (sendRequestToXapp(message, RIC_E2_RESET_RESP, rmrMessageBuffer) != 0) { + mdclog_write(MDCLOG_ERR, "RIC_E2_RESET_RESP message failed to send to xAPP"); } break; } @@ -1560,7 +1761,9 @@ void asnSuccsesfulMsg(E2AP_PDU_t *pdu, if (ie->value.present == RICcontrolAcknowledge_IEs__value_PR_RICrequestID) { message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_CONTROL_ACK; rmrMessageBuffer.sendMessage->state = 0; - rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID; +// rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID; + rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricInstanceID; + static unsigned char tx[32]; snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++); rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx)); @@ -1568,6 +1771,8 @@ void asnSuccsesfulMsg(E2AP_PDU_t *pdu, (unsigned char *)message.message.enodbName, strlen(message.message.enodbName)); + message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment(); + message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength); sendRmrMessage(rmrMessageBuffer, message); messageSent = true; } else { @@ -1581,66 +1786,12 @@ void asnSuccsesfulMsg(E2AP_PDU_t *pdu, break; } - case ProcedureCode_id_RICindication: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICindication %s", message.message.enodbName); - } - for (auto i = 0; i < pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.count; i++) { - auto messageSent = false; - RICindication_IEs_t *ie = pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.array[i]; - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id); - } - if (ie->id == ProtocolIE_ID_id_RICrequestID) { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id); - } - if (ie->value.present == RICindication_IEs__value_PR_RICrequestID) { - static unsigned char tx[32]; - message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_INDICATION; - snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++); - rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx)); - rmr_bytes2meid(rmrMessageBuffer.sendMessage, - (unsigned char *)message.message.enodbName, - strlen(message.message.enodbName)); - rmrMessageBuffer.sendMessage->state = 0; - rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID; - if (mdclog_level_get() >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "RIC sub id = %d, message type = %d", - rmrMessageBuffer.sendMessage->sub_id, - rmrMessageBuffer.sendMessage->mtype); - } - sendRmrMessage(rmrMessageBuffer, message); - messageSent = true; - } else { - mdclog_write(MDCLOG_ERR, "RIC request id missing illigal request"); - } - } - if (messageSent) { - break; - } - } - break; - } - case ProcedureCode_id_RICserviceQuery: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICserviceQuery %s", message.message.enodbName); - } - break; - } - case ProcedureCode_id_RICserviceUpdate: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICserviceUpdate %s", message.message.enodbName); - } - if (sendRequestToXapp(message, RIC_SERVICE_UPDATE, rmrMessageBuffer) != 0) { - mdclog_write(MDCLOG_ERR, "RIC_SERVICE_UPDATE message failed to send to xAPP"); - } - break; - } case ProcedureCode_id_RICsubscription: { if (logLevel >= MDCLOG_DEBUG) { mdclog_write(MDCLOG_DEBUG, "Got RICsubscription %s", message.message.enodbName); } + message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment(); + message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength); if (sendRequestToXapp(message, RIC_SUB_RESP, rmrMessageBuffer) != 0) { mdclog_write(MDCLOG_ERR, "Subscription successful message failed to send to xAPP"); } @@ -1650,6 +1801,8 @@ void asnSuccsesfulMsg(E2AP_PDU_t *pdu, if (logLevel >= MDCLOG_DEBUG) { mdclog_write(MDCLOG_DEBUG, "Got RICsubscriptionDelete %s", message.message.enodbName); } + message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment(); + message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength); if (sendRequestToXapp(message, RIC_SUB_DEL_RESP, rmrMessageBuffer) != 0) { mdclog_write(MDCLOG_ERR, "Subscription delete successful message failed to send to xAPP"); } @@ -1681,30 +1834,6 @@ void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu, mdclog_write(MDCLOG_INFO, "Unsuccessful Outcome %ld", procedureCode); } switch (procedureCode) { - case ProcedureCode_id_E2setup: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got E2setup\n"); - } - break; - } - case ProcedureCode_id_ErrorIndication: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got ErrorIndication %s", message.message.enodbName); - } - if (sendRequestToXapp(message, RIC_ERROR_INDICATION, rmrMessageBuffer) != 0) { - mdclog_write(MDCLOG_ERR, "RIC_ERROR_INDICATION failed to send to xAPP"); - } - break; - } - case ProcedureCode_id_Reset: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got Reset %s", message.message.enodbName); - } - if (sendRequestToXapp(message, RIC_X2_RESET, rmrMessageBuffer) != 0) { - mdclog_write(MDCLOG_ERR, "RIC_X2_RESET message failed to send to xAPP"); - } - break; - } case ProcedureCode_id_RICcontrol: { if (logLevel >= MDCLOG_DEBUG) { mdclog_write(MDCLOG_DEBUG, "Got RICcontrol %s", message.message.enodbName); @@ -1723,12 +1852,15 @@ void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu, if (ie->value.present == RICcontrolFailure_IEs__value_PR_RICrequestID) { message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_CONTROL_FAILURE; rmrMessageBuffer.sendMessage->state = 0; - rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID; +// rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricRequestorID; + rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricInstanceID; static unsigned char tx[32]; snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++); rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx)); rmr_bytes2meid(rmrMessageBuffer.sendMessage, (unsigned char *) message.message.enodbName, strlen(message.message.enodbName)); + message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment(); + message.peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength); sendRmrMessage(rmrMessageBuffer, message); messageSent = true; } else { @@ -1741,66 +1873,12 @@ void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu, } break; } - case ProcedureCode_id_RICindication: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICindication %s", message.message.enodbName); - } - for (auto i = 0; i < pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.count; i++) { - auto messageSent = false; - RICindication_IEs_t *ie = pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.array[i]; - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id); - } - if (ie->id == ProtocolIE_ID_id_RICrequestID) { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id); - } - if (ie->value.present == RICindication_IEs__value_PR_RICrequestID) { - static unsigned char tx[32]; - message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_INDICATION; - snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++); - rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx)); - rmr_bytes2meid(rmrMessageBuffer.sendMessage, - (unsigned char *)message.message.enodbName, - strlen(message.message.enodbName)); - rmrMessageBuffer.sendMessage->state = 0; - rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID; - if (mdclog_level_get() >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "RIC sub id = %d, message type = %d", - rmrMessageBuffer.sendMessage->sub_id, - rmrMessageBuffer.sendMessage->mtype); - } - sendRmrMessage(rmrMessageBuffer, message); - messageSent = true; - } else { - mdclog_write(MDCLOG_ERR, "RIC request id missing illigal request"); - } - } - if (messageSent) { - break; - } - } - break; - } - case ProcedureCode_id_RICserviceQuery: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICserviceQuery %s", message.message.enodbName); - } - break; - } - case ProcedureCode_id_RICserviceUpdate: { - if (logLevel >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "Got RICserviceUpdate %s", message.message.enodbName); - } - if (sendRequestToXapp(message, RIC_SERVICE_UPDATE, rmrMessageBuffer) != 0) { - mdclog_write(MDCLOG_ERR, "RIC_SERVICE_UPDATE message failed to send to xAPP"); - } - break; - } case ProcedureCode_id_RICsubscription: { if (logLevel >= MDCLOG_DEBUG) { mdclog_write(MDCLOG_DEBUG, "Got RICsubscription %s", message.message.enodbName); } + message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment(); + message.peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength); if (sendRequestToXapp(message, RIC_SUB_FAILURE, rmrMessageBuffer) != 0) { mdclog_write(MDCLOG_ERR, "Subscription unsuccessful message failed to send to xAPP"); } @@ -1810,7 +1888,9 @@ void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu, if (logLevel >= MDCLOG_DEBUG) { mdclog_write(MDCLOG_DEBUG, "Got RICsubscriptionDelete %s", message.message.enodbName); } - if (sendRequestToXapp(message, RIC_SUB_DEL_FAILURE, rmrMessageBuffer) != 0) { + message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment(); + message.peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength); + if (sendRequestToXapp(message, RIC_SUB_FAILURE, rmrMessageBuffer) != 0) { mdclog_write(MDCLOG_ERR, "Subscription Delete unsuccessful message failed to send to xAPP"); } break; @@ -1849,10 +1929,13 @@ int sendRequestToXapp(ReportingMessages_t &message, return rc; } - +/** + * + * @param pSctpParams + */ void getRmrContext(sctp_params_t &pSctpParams) { pSctpParams.rmrCtx = nullptr; - pSctpParams.rmrCtx = rmr_init(pSctpParams.rmrAddress, RMR_MAX_RCV_BYTES, RMRFL_NONE); + pSctpParams.rmrCtx = rmr_init(pSctpParams.rmrAddress, RECEIVE_XAPP_BUFFER_SIZE, RMRFL_NONE); if (pSctpParams.rmrCtx == nullptr) { mdclog_write(MDCLOG_ERR, "Failed to initialize RMR"); return; @@ -1893,28 +1976,43 @@ void getRmrContext(sctp_params_t &pSctpParams) { } } -int BuildPERSetupResponseMessaeFromXML(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) { - E2AP_PDU_t *pdu; +/** + * + * @param message + * @param rmrMessageBuffer + * @return + */ +int PER_FromXML(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) { + E2AP_PDU_t *pdu = nullptr; + + if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "got xml Format data from xApp of size %d is:%s", + rmrMessageBuffer.rcvMessage->len, rmrMessageBuffer.rcvMessage->payload); + } auto rval = asn_decode(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, (void **) &pdu, rmrMessageBuffer.rcvMessage->payload, rmrMessageBuffer.rcvMessage->len); if (rval.code != RC_OK) { - mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) E2AP PDU from E2MGR : %s", + mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) setup response from E2MGR : %s", rval.code, message.message.enodbName); return -1; } - auto er = asn_encode_to_buffer(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, pdu, - rmrMessageBuffer.rcvMessage->payload, rmrMessageBuffer.rcvMessage->len); + int buff_size = RECEIVE_XAPP_BUFFER_SIZE; + auto er = asn_encode_to_buffer(nullptr, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2AP_PDU, pdu, + rmrMessageBuffer.rcvMessage->payload, buff_size); if (er.encoded == -1) { mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno)); return -1; - } else if (er.encoded > (ssize_t)rmrMessageBuffer.rcvMessage->len) { - mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s", + } else if (er.encoded > (ssize_t)buff_size) { + mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s, at %s line %d", (int)rmrMessageBuffer.rcvMessage->len, - asn_DEF_E2AP_PDU.name); + asn_DEF_E2AP_PDU.name, + __func__, + __LINE__); return -1; } + rmrMessageBuffer.rcvMessage->len = er.encoded; return 0; } @@ -1955,12 +2053,27 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, return -1; } rmr_get_meid(rmrMessageBuffer.rcvMessage, (unsigned char *)message.message.enodbName); + message.peerInfo = (ConnectedCU_t *) sctpMap->find(message.message.enodbName); + if (message.peerInfo == nullptr) { + auto type = rmrMessageBuffer.rcvMessage->mtype; + switch (type) { + case RIC_SCTP_CLEAR_ALL: + case E2_TERM_KEEP_ALIVE_REQ: + case RIC_HEALTH_CHECK_REQ: + break; + default: + mdclog_write(MDCLOG_ERR, "Failed to send message no CU entry %s", message.message.enodbName); + return -1; + } + } + switch (rmrMessageBuffer.rcvMessage->mtype) { case RIC_E2_SETUP_RESP : { - if (BuildPERSetupResponseMessaeFromXML(message, rmrMessageBuffer) != 0) { + if (PER_FromXML(message, rmrMessageBuffer) != 0) { break; } - + message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment(); + message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_SETUP_RESP"); return -6; @@ -1968,9 +2081,11 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, break; } case RIC_E2_SETUP_FAILURE : { - if (BuildPERSetupResponseMessaeFromXML(message, rmrMessageBuffer) != 0) { + if (PER_FromXML(message, rmrMessageBuffer) != 0) { break; } + message.peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment(); + message.peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_SETUP_FAILURE"); return -6; @@ -1978,6 +2093,8 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, break; } case RIC_ERROR_INDICATION: { + message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment(); + message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { mdclog_write(MDCLOG_ERR, "Failed to send RIC_ERROR_INDICATION"); return -6; @@ -1985,6 +2102,8 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, break; } case RIC_SUB_REQ: { + message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment(); + message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { mdclog_write(MDCLOG_ERR, "Failed to send RIC_SUB_REQ"); return -6; @@ -1992,6 +2111,8 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, break; } case RIC_SUB_DEL_REQ: { + message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment(); + message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { mdclog_write(MDCLOG_ERR, "Failed to send RIC_SUB_DEL_REQ"); return -6; @@ -1999,6 +2120,8 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, break; } case RIC_CONTROL_REQ: { + message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment(); + message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { mdclog_write(MDCLOG_ERR, "Failed to send RIC_CONTROL_REQ"); return -6; @@ -2006,6 +2129,11 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, break; } case RIC_SERVICE_QUERY: { + if (PER_FromXML(message, rmrMessageBuffer) != 0) { + break; + } + message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment(); + message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_QUERY"); return -6; @@ -2013,6 +2141,11 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, break; } case RIC_SERVICE_UPDATE_ACK: { + if (PER_FromXML(message, rmrMessageBuffer) != 0) { + break; + } + message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(); + message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_UPDATE_ACK"); return -6; @@ -2020,22 +2153,37 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, break; } case RIC_SERVICE_UPDATE_FAILURE: { + if (PER_FromXML(message, rmrMessageBuffer) != 0) { + break; + } + message.peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(); + message.peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_UPDATE_FAILURE"); return -6; } break; } - case RIC_X2_RESET: { + case RIC_E2_RESET_REQ: { + if (PER_FromXML(message, rmrMessageBuffer) != 0) { + break; + } + message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment(); + message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { - mdclog_write(MDCLOG_ERR, "Failed to send RIC_X2_RESET"); + mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_RESET"); return -6; } break; } - case RIC_X2_RESET_RESP: { + case RIC_E2_RESET_RESP: { + if (PER_FromXML(message, rmrMessageBuffer) != 0) { + break; + } + message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment(); + message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len); if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) { - mdclog_write(MDCLOG_ERR, "Failed to send RIC_X2_RESET_RESP"); + mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_RESET_RESP"); return -6; } break; @@ -2098,6 +2246,31 @@ int receiveXappMessages(Sctp_Map_t *sctpMap, break; } + case RIC_HEALTH_CHECK_REQ: { + // send message back + rmr_bytes2payload(rmrMessageBuffer.rcvMessage, + (unsigned char *)"OK", + 2); + rmrMessageBuffer.rcvMessage->mtype = RIC_HEALTH_CHECK_RESP; + rmrMessageBuffer.rcvMessage->state = 0; + static unsigned char tx[32]; + auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++); + rmr_bytes2xact(rmrMessageBuffer.rcvMessage, tx, txLen); + rmrMessageBuffer.rcvMessage = rmr_rts_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.rcvMessage); + //rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage); + if (rmrMessageBuffer.rcvMessage == nullptr) { + rmrMessageBuffer.rcvMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE); + mdclog_write(MDCLOG_ERR, "Failed to send RIC_HEALTH_CHECK_RESP RMR message returned NULL"); + } else if (rmrMessageBuffer.rcvMessage->state != 0) { + mdclog_write(MDCLOG_ERR, "Failed to send RIC_HEALTH_CHECK_RESP, on RMR state = %d ( %s)", + rmrMessageBuffer.rcvMessage->state, translateRmrErrorMessages(rmrMessageBuffer.rcvMessage->state).c_str()); + } else if (mdclog_level_get() >= MDCLOG_DEBUG) { + mdclog_write(MDCLOG_DEBUG, "Got RIC_HEALTH_CHECK_REQ Request send : OK"); + } + + break; + } + default: mdclog_write(MDCLOG_WARN, "Message Type : %d is not seported", rmrMessageBuffer.rcvMessage->mtype); message.message.asndata = rmrMessageBuffer.rcvMessage->payload; @@ -2151,48 +2324,12 @@ int sendMessagetoCu(Sctp_Map_t *sctpMap, RmrMessagesBuffer_t &messageBuffer, ReportingMessages_t &message, int failedMesgId) { - auto *peerInfo = (ConnectedCU_t *) sctpMap->find(message.message.enodbName); - if (peerInfo == nullptr) { - if (failedMesgId != 0) { - sendFailedSendingMessagetoXapp(messageBuffer, message, failedMesgId); - } else { - mdclog_write(MDCLOG_ERR, "Failed to send message no CU entry %s", message.message.enodbName); - } - return -1; - } - // get the FD message.message.messageType = messageBuffer.rcvMessage->mtype; - auto rc = sendSctpMsg(peerInfo, message, sctpMap); + auto rc = sendSctpMsg(message.peerInfo, message, sctpMap); return rc; } -/** - * - * @param rmrCtx the rmr context to send and receive - * @param msg the msg we got fromxApp - * @param metaData data from xApp in ordered struct - * @param failedMesgId the return message type error - */ -void -sendFailedSendingMessagetoXapp(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message, int failedMesgId) { - rmr_mbuf_t *msg = rmrMessageBuffer.sendMessage; - msg->len = snprintf((char *) msg->payload, 200, "the gNb/eNode name %s not found", - message.message.enodbName); - if (mdclog_level_get() >= MDCLOG_INFO) { - mdclog_write(MDCLOG_INFO, "%s", msg->payload); - } - msg->mtype = failedMesgId; - msg->state = 0; - - static unsigned char tx[32]; - snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++); - rmr_bytes2xact(msg, tx, strlen((const char *) tx)); - - sendRmrMessage(rmrMessageBuffer, message); -} - - /** * @@ -2337,7 +2474,7 @@ void buildJsonMessage(ReportingMessages_t &message) { message.base64Data, message.outLen); if (mdclog_level_get() >= MDCLOG_DEBUG) { - mdclog_write(MDCLOG_DEBUG, "asn data length = %d, base64 message length = %d ", + mdclog_write(MDCLOG_DEBUG, "Tracing: ASN length = %d, base64 message length = %d ", (int) message.message.asnLength, (int) message.outLen); }