src/afb-supervision.c

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