src/afb-supervision.c

   1 /*
   2  * Copyright (C) 2016-2019 "IoT.bzh"
   3  * Author José Bollo <jose.bollo@iot.bzh>
   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 #if defined(WITH_SUPERVISION)
  19
  20 #define _GNU_SOURCE
  21
  22 #include <string.h>
  23 #include <errno.h>
  24 #include <signal.h>
  25 #include <unistd.h>
  26 #include <fcntl.h>
  27 #include <pthread.h>
  28 #include <sys/types.h>
  29 #include <sys/socket.h>
  30 #include <sys/un.h>
  31
  32 #include <json-c/json.h>
  33
  34 #define AFB_BINDING_VERSION 3
  35 #include <afb/afb-binding.h>
  36
  37 #include "afb-cred.h"
  38 #include "afb-api.h"
  39 #include "afb-apiset.h"
  40 #include "afb-api-so-v2.h"
  41 #include "afb-xreq.h"
  42 #include "afb-trace.h"
  43 #include "afb-session.h"
  44 #include "afb-config.h"
  45 #include "afb-supervision.h"
  46 #include "afs-supervision.h"
  47 #include "afb-stub-ws.h"
  48 #include "afb-debug.h"
  49 #include "afb-fdev.h"
  50 #include "verbose.h"
  51 #include "wrap-json.h"
  52 #include "jobs.h"
  53
  54 /* api and apiset name */
  55 static const char supervision_apiname[] = AFS_SUPERVISION_APINAME;
  56 #if WITH_AFB_TRACE
  57 static const char supervisor_apiname[] = AFS_SUPERVISOR_APINAME;
  58 #endif
  59
  60 /* path of the supervision socket */
  61 static const char supervisor_socket_path[] = AFS_SUPERVISION_SOCKET;
  62
  63 /* mutual exclusion */
  64 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  65
  66 /* the standard apiset */
  67 static struct {
  68         struct afb_apiset *apiset;
  69         struct json_object *config;
  70 } global;
  71
  72 /* the supervision apiset (not exported) */
  73 static struct afb_apiset *supervision_apiset;
  74
  75 /* local api implementation */
  76 static void on_supervision_call(void *closure, struct afb_xreq *xreq);
  77 static struct afb_api_itf supervision_api_itf =
  78 {
  79         .call = on_supervision_call
  80 };
  81
  82 /* the supervisor link */
  83 static struct afb_stub_ws *supervisor;
  84
  85 #if WITH_AFB_TRACE
  86 /* the trace api */
  87 static struct afb_trace *trace;
  88 #endif
  89
  90 /* open the socket */
  91 static int open_supervisor_socket(const char *path)
  92 {
  93         int fd, rc;
  94         struct sockaddr_un addr;
  95         size_t length;
  96
  97         /* check path length */
  98         length = strlen(path);
  99         if (length >= 108) {
 100                 errno = ENAMETOOLONG;
 101                 return -1;
 102         }
 103
 104         /* create the unix socket */
 105         fd = socket(AF_UNIX, SOCK_STREAM, 0);
 106         if (fd < 0)
 107                 return fd;
 108
 109         /* prepare the connection address */
 110         memset(&addr, 0, sizeof addr);
 111         addr.sun_family = AF_UNIX;
 112         strcpy(addr.sun_path, path);
 113         if (addr.sun_path[0] == '@')
 114                 addr.sun_path[0] = 0; /* implement abstract sockets */
 115
 116         /* connect the socket */
 117         rc = connect(fd, (struct sockaddr *) &addr, (socklen_t)(sizeof addr));
 118         if (rc < 0) {
 119                 close(fd);
 120                 return rc;
 121         }
 122         return fd;
 123 }
 124
 125 static void disconnect_supervisor()
 126 {
 127         struct afb_stub_ws *s;
 128
 129         INFO("Disconnecting supervision");
 130         s = __atomic_exchange_n(&supervisor, NULL, __ATOMIC_RELAXED);
 131         if (s)
 132                 afb_stub_ws_unref(s);
 133
 134 #if WITH_AFB_TRACE
 135         struct afb_trace *t = __atomic_exchange_n(&trace, NULL, __ATOMIC_RELAXED);
 136         if (t)
 137                 afb_trace_unref(t);
 138 #endif
 139 }
 140
 141 static void on_supervisor_hangup(struct afb_stub_ws *s)
 142 {
 143         if (s && s == supervisor)
 144                 disconnect_supervisor();
 145 }
 146
 147 /* try to connect to supervisor */
 148 static void try_connect_supervisor()
 149 {
 150         int fd;
 151         ssize_t srd;
 152         struct afs_supervision_initiator initiator;
 153         struct fdev *fdev;
 154
 155         /* get the mutex */
 156         pthread_mutex_lock(&mutex);
 157
 158         /* needs to connect? */
 159         if (supervisor || !supervision_apiset)
 160                 goto end;
 161
 162         /* check existing path */
 163         if (supervisor_socket_path[0] != '@' && access(supervisor_socket_path, F_OK)) {
 164                 NOTICE("Can't acces socket path %s: %m", supervisor_socket_path);
 165                 goto end;
 166         }
 167
 168         /* socket connection */
 169         fd = open_supervisor_socket(supervisor_socket_path);
 170         if (fd < 0) {
 171                 NOTICE("Can't connect supervision socket to %s: %m", supervisor_socket_path);
 172                 goto end;
 173         }
 174
 175         /* negotiation */
 176         do { srd = read(fd, &initiator, sizeof initiator); } while(srd < 0 && errno == EINTR);
 177         if (srd < 0) {
 178                 NOTICE("Can't read supervisor %s: %m", supervisor_socket_path);
 179                 goto end2;
 180         }
 181         if ((size_t)srd != sizeof initiator) {
 182                 ERROR("When reading supervisor %s: %m", supervisor_socket_path);
 183                 goto end2;
 184         }
 185         if (strnlen(initiator.interface, sizeof initiator.interface) == sizeof initiator.interface) {
 186                 ERROR("Bad interface of supervisor %s", supervisor_socket_path);
 187                 goto end2;
 188         }
 189         if (strcmp(initiator.interface, AFS_SUPERVISION_INTERFACE_1)) {
 190                 ERROR("Unknown interface %s for supervisor %s", initiator.interface, supervisor_socket_path);
 191                 goto end2;
 192         }
 193         if (strnlen(initiator.extra, sizeof initiator.extra) == sizeof initiator.extra) {
 194                 ERROR("Bad extra of supervisor %s", supervisor_socket_path);
 195                 goto end2;
 196         }
 197
 198         /* interprets extras */
 199         if (!strcmp(initiator.extra, "CLOSE")) {
 200                 INFO("Supervisor asks to CLOSE");
 201                 goto end2;
 202         }
 203         if (!strcmp(initiator.extra, "WAIT")) {
 204                 afb_debug_wait("supervisor");
 205         }
 206         if (!strcmp(initiator.extra, "BREAK")) {
 207                 afb_debug_break("supervisor");
 208         }
 209
 210         /* make the supervisor link */
 211         fdev = afb_fdev_create(fd);
 212         if (!fdev) {
 213                 ERROR("Creation of fdev failed: %m");
 214                 goto end2;
 215         }
 216         supervisor = afb_stub_ws_create_server(fdev, supervision_apiname, supervision_apiset);
 217         if (!supervisor) {
 218                 ERROR("Creation of supervisor failed: %m");
 219                 goto end;
 220         }
 221         afb_stub_ws_set_on_hangup(supervisor, on_supervisor_hangup);
 222
 223         /* successful termination */
 224         goto end;
 225
 226 end2:
 227         close(fd);
 228 end:
 229         pthread_mutex_unlock(&mutex);
 230 }
 231
 232 static void try_connect_supervisor_job(int signum, void *args)
 233 {
 234         INFO("Try to connect supervisor after SIGHUP");
 235         try_connect_supervisor();
 236 }
 237
 238 static void on_sighup(int signum)
 239 {
 240         INFO("Supervision received a SIGHUP");
 241         jobs_queue(NULL, 0, try_connect_supervisor_job, NULL);
 242 }
 243
 244 /**
 245  * initialize the supervision
 246  */
 247 int afb_supervision_init(struct afb_apiset *apiset, struct json_object *config)
 248 {
 249         int rc;
 250         struct sigaction sa;
 251
 252         /* don't reinit */
 253         if (supervision_apiset)
 254                 return 0;
 255
 256         /* create the apiset */
 257         supervision_apiset = afb_apiset_create(supervision_apiname, 0);
 258         if (!supervision_apiset) {
 259                 ERROR("Can't create supervision's apiset");
 260                 return -1;
 261         }
 262
 263         /* init the apiset */
 264         rc = afb_apiset_add(supervision_apiset, supervision_apiname,
 265                         (struct afb_api_item){ .itf = &supervision_api_itf, .closure = NULL});
 266         if (rc < 0) {
 267                 ERROR("Can't create supervision's apiset: %m");
 268                 afb_apiset_unref(supervision_apiset);
 269                 supervision_apiset = NULL;
 270                 return rc;
 271         }
 272
 273         /* init the globals */
 274         global.apiset = apiset;
 275         global.config = config;
 276
 277         /* get SIGHUP */
 278         memset(&sa, 0, sizeof sa);
 279         sa.sa_handler = on_sighup;
 280         rc = sigaction(SIGHUP, &sa, NULL);
 281         if (rc < 0)
 282                 ERROR("Can't connect supervision to SIGHUP: %m");
 283
 284         /* connect to supervision */
 285         try_connect_supervisor();
 286         return 0;
 287 }
 288
 289 /******************************************************************************
 290 **** Implementation monitoring verbs
 291 ******************************************************************************/
 292 static void slist(void *closure, struct afb_session *session)
 293 {
 294         struct json_object *list = closure;
 295         struct json_object *item;
 296
 297         wrap_json_pack(&item, "{ss}", "token", afb_session_token(session));
 298         json_object_object_add(list, afb_session_uuid(session), item);
 299 }
 300
 301 /******************************************************************************
 302 **** Implementation monitoring verbs
 303 ******************************************************************************/
 304
 305 static const char *verbs[] = {
 306         "break", "config", "do", "exit", "sclose", "slist", "trace", "wait" };
 307 enum  {  Break ,  Config ,  Do ,  Exit ,  Sclose ,  Slist ,  Trace ,  Wait  };
 308
 309 static void on_supervision_call(void *closure, struct afb_xreq *xreq)
 310 {
 311         int i;
 312         struct json_object *args, *sub, *list;
 313         const char *api, *verb, *uuid;
 314         struct afb_session *session;
 315         const struct afb_api_item *xapi;
 316         afb_req_t req;
 317 #if WITH_AFB_TRACE
 318         struct json_object *drop, *add;
 319         int rc;
 320 #endif
 321
 322         /* search the verb */
 323         i = (int)(sizeof verbs / sizeof *verbs);
 324         while(--i >= 0 && strcasecmp(verbs[i], xreq->request.called_verb));
 325         if (i < 0) {
 326                 afb_xreq_reply_unknown_verb(xreq);
 327                 return;
 328         }
 329
 330         /* process */
 331         args = afb_xreq_json(xreq);
 332         switch(i) {
 333         case Exit:
 334                 i = 0;
 335                 if (wrap_json_unpack(args, "i", &i))
 336                         wrap_json_unpack(args, "{si}", "code", &i);
 337                 ERROR("exiting from supervision with code %d -> %d", i, i & 127);
 338                 exit(i & 127);
 339                 break;
 340         case Sclose:
 341                 uuid = NULL;
 342                 if (wrap_json_unpack(args, "s", &uuid))
 343                         wrap_json_unpack(args, "{ss}", "uuid", &uuid);
 344                 if (!uuid)
 345                         afb_xreq_reply(xreq, NULL, "invalid", NULL);
 346                 else {
 347                         session = afb_session_search(uuid);
 348                         if (!session)
 349                                 afb_xreq_reply(xreq, NULL, "not-found", NULL);
 350                         else {
 351                                 afb_session_close(session);
 352                                 afb_session_unref(session);
 353                                 afb_session_purge();
 354                                 afb_xreq_reply(xreq, NULL, NULL, NULL);
 355                         }
 356                 }
 357                 break;
 358         case Slist:
 359                 list = json_object_new_object();
 360                 afb_session_foreach(slist, list);
 361                 afb_xreq_reply(xreq, list, NULL, NULL);
 362                 break;
 363         case Config:
 364                 afb_xreq_reply(xreq, json_object_get(global.config), NULL, NULL);
 365                 break;
 366         case Trace:
 367                 req = xreq_to_req_x2(xreq);
 368 #if WITH_AFB_TRACE
 369                 if (!trace)
 370                         trace = afb_trace_create(supervisor_apiname, NULL /* not bound to any session */);
 371
 372                 add = drop = NULL;
 373                 wrap_json_unpack(args, "{s?o s?o}", "add", &add, "drop", &drop);
 374                 if (add) {
 375                         rc = afb_trace_add(req, add, trace);
 376                         if (rc)
 377                                 return;
 378                 }
 379                 if (drop) {
 380                         rc = afb_trace_drop(req, drop, trace);
 381                         if (rc)
 382                                 return;
 383                 }
 384                 afb_req_success(req, NULL, NULL);
 385 #else
 386                 afb_req_reply(req, NULL, "not-available", NULL);
 387 #endif
 388                 break;
 389         case Do:
 390                 sub = NULL;
 391                 if (wrap_json_unpack(args, "{ss ss s?o*}", "api", &api, "verb", &verb, "args", &sub))
 392                         afb_xreq_reply(xreq, NULL, "error", "bad request");
 393                 else {
 394                         xapi = afb_apiset_lookup_started(global.apiset, api, 1);
 395                         if (!xapi)
 396                                 afb_xreq_reply_unknown_api(xreq);
 397                         else {
 398                                 afb_cred_unref(xreq->cred);
 399                                 xreq->cred = NULL;
 400                                 xreq->request.called_api = api;
 401                                 xreq->request.called_verb = verb;
 402                                 xreq->json = json_object_get(sub);
 403                                 xapi->itf->call(xapi->closure, xreq);
 404                                 json_object_put(args);
 405                         }
 406                 }
 407                 break;
 408         case Wait:
 409                 afb_xreq_reply(xreq, NULL, NULL, NULL);
 410                 afb_debug_wait("supervisor");
 411                 break;
 412         case Break:
 413                 afb_xreq_reply(xreq, NULL, NULL, NULL);
 414                 afb_debug_break("supervisor");
 415                 break;
 416         }
 417 }
 418
 419 #endif