src/afb-supervision.c

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