src/afb-supervision.c

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