RIC-E2-TERMINATION/sctpThread.h

   1 /*
   2  * Copyright 2019 AT&T Intellectual Property
   3  * Copyright 2019 Nokia
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16 */
  17
  18 /*
  19  * This source code is part of the near-RT RIC (RAN Intelligent Controller)
  20  * platform project (RICP).
  21  */
  22
  23 #ifndef X2_SCTP_THREAD_H
  24 #define X2_SCTP_THREAD_H
  25
  26 #include <algorithm>
  27
  28 #include <cstdio>
  29 #include <cerrno>
  30 #include <cstdlib>
  31 #include <cstring>
  32 #include <random>
  33 #include <sys/socket.h>
  34 #include <arpa/inet.h>
  35 #include <netinet/in_systm.h>
  36 #include <netinet/in.h>
  37 #include <netinet/ip.h>
  38 #include <netinet/ip_icmp.h>
  39 #include <netinet/sctp.h>
  40 #include <thread>
  41 #include <atomic>
  42 #include <sys/param.h>
  43 #include <sys/file.h>
  44 #include <sys/types.h>
  45 #include <ifaddrs.h>
  46 #include <ctime>
  47 #include <netdb.h>
  48 #include <sys/epoll.h>
  49 #include <mutex>
  50 #include <shared_mutex>
  51 #include <iterator>
  52 #include <map>
  53 #include <sys/inotify.h>
  54 #include <csignal>
  55 #include <future>
  56
  57 #include <rmr/rmr.h>
  58 #include <rmr/RIC_message_types.h>
  59 #include <mdclog/mdclog.h>
  60 #include <functional>
  61 #include <iostream>
  62
  63 #include <boost/algorithm/string/predicate.hpp>
  64 #include <boost/lexical_cast.hpp>
  65 #include <boost/move/utility.hpp>
  66 #include <boost/log/sources/logger.hpp>
  67 #include <boost/log/sources/record_ostream.hpp>
  68 #include <boost/log/sources/global_logger_storage.hpp>
  69 #include <boost/log/utility/setup/file.hpp>
  70 #include <boost/log/utility/setup/common_attributes.hpp>
  71 #include <boost/filesystem.hpp>
  72
  73 #include <mdclog/mdclog.h>
  74
  75 #include "oranE2/E2AP-PDU.h"
  76 #include "oranE2/ProtocolIE-Container.h"
  77 #include "oranE2/InitiatingMessage.h"
  78 #include "oranE2/SuccessfulOutcome.h"
  79 #include "oranE2/UnsuccessfulOutcome.h"
  80 #include "oranE2/ProtocolIE-Container.h"
  81 #include "oranE2/ProtocolIE-Field.h"
  82 #include "oranE2/GlobalE2node-gNB-ID.h"
  83 #include "oranE2/GlobalE2node-en-gNB-ID.h"
  84 #include "oranE2/GlobalE2node-ng-eNB-ID.h"
  85 #include "oranE2/GlobalE2node-eNB-ID.h"
  86
  87 #include "cxxopts.hpp"
  88 //#include "config-cpp/include/config-cpp/config-cpp.h"
  89 #include <zlib.h>
  90 #include <prometheus/counter.h>
  91 #include <prometheus/exposer.h>
  92 #include <prometheus/gateway.h>
  93 #include <prometheus/registry.h>
  94
  95 using namespace prometheus;
  96
  97 #include "mapWrapper.h"
  98
  99 #include "base64.h"
 100
 101 #include "ReadConfigFile.h"
 102
 103 using namespace std;
 104 namespace logging = boost::log;
 105 namespace src = boost::log::sources;
 106 namespace keywords = boost::log::keywords;
 107 namespace sinks = boost::log::sinks;
 108 namespace posix_time = boost::posix_time;
 109 namespace expr = boost::log::expressions;
 110
 111 #define SRC_PORT 36422
 112 #define SA      struct sockaddr
 113 #define MAX_ENODB_NAME_SIZE 64
 114
 115 #define MAXEVENTS 128
 116
 117 #define RECEIVE_SCTP_BUFFER_SIZE (256 * 1024)
 118 #define RECEIVE_XAPP_BUFFER_SIZE RECEIVE_SCTP_BUFFER_SIZE
 119
 120 typedef mapWrapper Sctp_Map_t;
 121
 122
 123
 124 #define VOLUME_URL_SIZE 256
 125 #define KA_MESSAGE_SIZE 2048
 126
 127 typedef struct sctp_params {
 128     int      epollTimeOut = -1;
 129     uint16_t rmrPort = 0;
 130     uint16_t sctpPort = SRC_PORT;
 131     int      epoll_fd = 0;
 132     int      listenFD = 0;
 133     int      rmrListenFd = 0;
 134     int      inotifyFD = 0;
 135     int      inotifyWD = 0;
 136     void     *rmrCtx = nullptr;
 137     Sctp_Map_t *sctpMap = nullptr;
 138     char      ka_message[KA_MESSAGE_SIZE] {};
 139     int       ka_message_length = 0;
 140     char       rmrAddress[256] {}; // "tcp:port number" "tcp:5566" listen to all address on port 5566
 141     char volume[VOLUME_URL_SIZE];
 142     string myIP {};
 143     string fqdn {};
 144     string podName {};
 145     string configFilePath {};
 146     string configFileName {};
 147     bool trace = true;
 148     shared_ptr<prometheus::Registry> prometheusRegistry;
 149     string prometheusPort {"8088"};
 150     Family<Counter> *prometheusFamily;
 151     Exposer *prometheusExposer = nullptr;
 152     Counter *e2tCounters[6][2][ProcedureCode_id_RICsubscriptionDeleteRequired + 1] {};
 153 } sctp_params_t;
 154
 155 // RAN to RIC
 156 #define IN_INITI 0 //INITIATING
 157 #define IN_SUCC 1 //SUCCESSFUL
 158 #define IN_UN_SUCC 2 //UN-Successful
 159
 160 // RIC To RAN
 161 #define OUT_INITI 3 //INITIATING
 162 #define OUT_SUCC 4 //SUCCESSFUL
 163 #define OUT_UN_SUCC 5 //UN-Successful
 164
 165 #define MSG_COUNTER 0
 166 #define BYTES_COUNTER 1
 167
 168 #define INVALID_STREAM_ID -1
 169
 170 typedef struct ConnectedCU {
 171     int fileDescriptor = 0;
 172     char hostName[NI_MAXHOST] {};
 173     char portNumber[NI_MAXSERV] {};
 174     char enodbName[MAX_ENODB_NAME_SIZE] {};
 175     char asnData[RECEIVE_SCTP_BUFFER_SIZE] {};
 176     size_t asnLength = 0;
 177     int mtype = 0;
 178     bool isConnected = false;
 179     bool gotSetup = false;
 180     sctp_params_t *sctpParams = nullptr;
 181     Counter *counters[6][2][ProcedureCode_id_RICsubscriptionDeleteRequired + 1] {};
 182     bool isSingleStream = false;
 183     int singleStreamId = 0;
 184 } ConnectedCU_t ;
 185
 186
 187 #define MAX_RMR_BUFF_ARRAY 32
 188 typedef struct RmrMessagesBuffer {
 189     char ka_message[KA_MESSAGE_SIZE] {};
 190     int  ka_message_len = 0;
 191     void *rmrCtx = nullptr;
 192     rmr_mbuf_t *sendMessage= nullptr;
 193     //rmr_mbuf_t *sendBufferedMessages[MAX_RMR_BUFF_ARRAY] {};
 194     rmr_mbuf_t *rcvMessage= nullptr;
 195     //rmr_mbuf_t *rcvBufferedMessages[MAX_RMR_BUFF_ARRAY] {};
 196 } RmrMessagesBuffer_t;
 197
 198 typedef struct formatedMessage {
 199     char enodbName[MAX_ENODB_NAME_SIZE];
 200     struct timespec time;
 201     int messageType;
 202     char direction;
 203     ssize_t asnLength;
 204     unsigned char *asndata;
 205 } FormatedMessage_t;
 206
 207 typedef struct ReportingMessages {
 208     FormatedMessage_t message {};
 209     ConnectedCU_t *peerInfo = nullptr;
 210     long outLen = 0;
 211     unsigned char base64Data[RECEIVE_SCTP_BUFFER_SIZE * 2] {};
 212     char buffer[RECEIVE_SCTP_BUFFER_SIZE * 8] {};
 213 } ReportingMessages_t;
 214
 215 cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &pSctpParams);
 216
 217 int buildInotify(sctp_params_t &sctpParams);
 218
 219 void handleTermInit(sctp_params_t &sctpParams);
 220
 221 void handleConfigChange(sctp_params_t *sctpParams);
 222
 223 void listener(sctp_params_t *params);
 224
 225 void sendTermInit(sctp_params_t &sctpParams);
 226
 227 int setSocketNoBlocking(int socket);
 228
 229 void handleEinprogressMessages(struct epoll_event &event,
 230                                ReportingMessages_t &message,
 231                                RmrMessagesBuffer_t &rmrMessageBuffer,
 232                                sctp_params_t *params);
 233
 234 void handlepoll_error(struct epoll_event &event,
 235                       ReportingMessages_t &message,
 236                       RmrMessagesBuffer_t &rmrMessageBuffer,
 237                       sctp_params_t *params);
 238
 239
 240 void cleanHashEntry(ConnectedCU_t *peerInfo, Sctp_Map_t *m);
 241
 242
 243 /**
 244  *
 245  * @param message
 246  * @param rmrMessageBuffer
 247  */
 248 void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer);
 249
 250 /**
 251  *
 252  * @param sctpMap
 253  * @param messageBuffer
 254  * @param message
 255  * @param failedMesgId
 256  * @return
 257  */
 258 int sendMessagetoCu(Sctp_Map_t *sctpMap,
 259                     RmrMessagesBuffer_t &messageBuffer,
 260                     ReportingMessages_t &message,
 261                     int failedMesgId);
 262
 263 void sendFailedSendingMessagetoXapp(RmrMessagesBuffer_t &rmrMessageBuffer,
 264                                     ReportingMessages_t &message,
 265                                     int failedMesgId);
 266
 267 int sendRequestToXapp(ReportingMessages_t &message,
 268                       int requestId,
 269                       RmrMessagesBuffer_t &rmrMmessageBuffer);
 270
 271 /**
 272  *
 273  * @param message
 274  * @param msgType
 275  * @param requestType
 276  * @param rmrMessageBuffer
 277  * @param sctpMap
 278  * @return
 279  */
 280 /*
 281 int sendResponseToXapp(ReportingMessages_t &message,
 282                        int msgType,
 283                        int requestType,
 284                        RmrMessagesBuffer_t &rmrMessageBuffer,
 285                        Sctp_Map_t *sctpMap);
 286 */
 287
 288 /**
 289  *
 290  * @param peerInfo
 291  * @param message
 292  * @param m
 293  * @return
 294  */
 295 int sendSctpMsg(ConnectedCU_t *peerInfo,
 296                 ReportingMessages_t &message,
 297                 Sctp_Map_t *m);
 298
 299 /**
 300  *
 301  * @param events
 302  * @param sctpMap
 303  * @param numOfMessages
 304  * @param rmrMessageBuffer
 305  * @param ts
 306  * @return
 307  */
 308 int receiveDataFromSctp(struct epoll_event *events,
 309                         Sctp_Map_t *sctpMap,
 310                         int &numOfMessages,
 311                         RmrMessagesBuffer_t &rmrMessageBuffer,
 312                         struct timespec &ts);
 313
 314 /**
 315  *
 316  * @param rmrAddress
 317  * @return
 318  */
 319 void getRmrContext(sctp_params_t &pSctpParams);
 320
 321 /**
 322  *
 323  * @param sctpMap
 324  * @param rmrMessageBuffer
 325  * @param ts
 326  * @return
 327  */
 328 int receiveXappMessages(Sctp_Map_t *sctpMap,
 329                         RmrMessagesBuffer_t &rmrMessageBuffer,
 330                         struct timespec &ts);
 331
 332 /**
 333  *
 334  * @param messageBuffer
 335  * @param failedMsgId
 336  * @param sctpMap
 337  * @return
 338  */
 339 int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messageBuffer,
 340                            ReportingMessages_t &message,
 341                            int failedMsgId,
 342                            Sctp_Map_t *sctpMap);
 343 /**
 344  *
 345  * @param pdu
 346  * @param message
 347  * @param rmrMessageBuffer
 348  */
 349 void asnInitiatingRequest(E2AP_PDU_t *pdu,
 350                           Sctp_Map_t *sctpMap,
 351                           ReportingMessages_t &message,
 352                           RmrMessagesBuffer_t &rmrMessageBuffer,int streamId);
 353 /**
 354  *
 355  * @param pdu
 356  * @param message
 357  * @param sctpMap
 358  * @param rmrMessageBuffer
 359  */
 360 void asnSuccessfulMsg(E2AP_PDU_t *pdu,
 361                       Sctp_Map_t *sctpMap,
 362                       ReportingMessages_t &message,
 363                       RmrMessagesBuffer_t &rmrMessageBuffer);
 364 /**
 365  *
 366  * @param pdu
 367  * @param message
 368  * @param sctpMap
 369  * @param rmrMessageBuffer
 370  */
 371 void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu,
 372                         Sctp_Map_t *sctpMap,
 373                         ReportingMessages_t &message,
 374                         RmrMessagesBuffer_t &rmrMessageBuffer);
 375
 376 /**
 377  *
 378  * @param rmrMessageBuffer
 379  * @param message
 380  * @return
 381  */
 382 int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message);
 383 /**
 384  *
 385  * @param epoll_fd
 386  * @param peerInfo
 387  * @param events
 388  * @param sctpMap
 389  * @param enodbName
 390  * @param msgType
 391  * @returnsrc::logger_mt& lg = my_logger::get();
 392  */
 393 int addToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType);
 394 /**
 395  *
 396  * @param epoll_fd
 397  * @param peerInfo
 398  * @param events
 399  * @param sctpMap
 400  * @param enodbName
 401  * @param msgType
 402  * @return
 403  */
 404 int modifyToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType);
 405
 406 /**
 407  *
 408  * @param message
 409  */
 410 void buildJsonMessage(ReportingMessages_t &message);
 411
 412 /**
 413  *
 414  *
 415  * @param state
 416  * @return
 417  */
 418 string translateRmrErrorMessages(int state);
 419
 420 int buildConfiguration(sctp_params_t &sctpParams);
 421 void startPrometheus(sctp_params_t &sctpParams);
 422 static int enable_log_change_notify(const char* fileName);
 423 static int register_log_change_notify(const char *fileName);
 424 static void * monitor_loglevel_change_handler(void* arg);
 425 void  update_mdc_log_level_severity(char* log_level);
 426 char* getinterfaceip();
 427 static char* parse_file(char* filename);
 428
 429
 430 static inline uint64_t rdtscp(uint32_t &aux) {
 431     uint64_t rax,rdx;
 432     asm volatile ("rdtscp\n" : "=a" (rax), "=d" (rdx), "=c" (aux) : :);
 433     return (rdx << (unsigned)32) + rax;
 434 }
 435 #ifndef RIC_SCTP_CONNECTION_FAILURE
 436 #define RIC_SCTP_CONNECTION_FAILURE  10080
 437 #endif
 438
 439 #ifdef UNIT_TEST
 440     #define FILE_DESCRIPTOR 53424 /*Dummy value for file descriptor only when UT is defined*/
 441 #endif
 442
 443 int buildListeningPort(sctp_params_t &sctpParams);
 444 void buildE2TPrometheusCounters(sctp_params_t &sctpParams);
 445
 446 int fetchStreamId(ConnectedCU_t *peerInfo, ReportingMessages_t &message);
 447 #endif //X2_SCTP_THREAD_H