plugin.c (12080B)
1 #define _GNU_SOURCE 2 #include <errno.h> 3 #include <netdb.h> 4 #include <signal.h> 5 #include <stdio.h> 6 #include <stdlib.h> 7 #include <string.h> 8 #include <strings.h> 9 #include <sys/socket.h> 10 #include <sys/types.h> 11 #include <sys/wait.h> 12 #include <time.h> 13 #include <unistd.h> 14 15 #include "finwo/resp.h" 16 #include "rxi/log.h" 17 18 #include "config/daemon.h" 19 #include "dataplane.h" 20 #include "dataplane/plugin.h" 21 22 #define PLUGIN_TIMEOUT_S 5 23 #define BACKOFF_MIN_MS 500 24 #define BACKOFF_MAX_MS 30000 25 #define MAX_CONFIG_ARGV 32 26 27 static struct dp_plugin *plugins = NULL; 28 29 static void strlist_free(char ***list, size_t *n) { 30 for (size_t i = 0; i < *n; i++) free((*list)[i]); 31 free(*list); 32 *list = NULL; 33 *n = 0; 34 } 35 36 static int strlist_add(char ***list, size_t *n, const char *s) { 37 char **g = realloc(*list, sizeof(char *) * (*n + 1)); 38 if (!g) return -1; 39 *list = g; 40 (*list)[*n] = strdup(s); 41 if (!(*list)[*n]) return -1; 42 (*n)++; 43 return 0; 44 } 45 46 static int strlist_has(char **list, size_t n, const char *s) { 47 for (size_t i = 0; i < n; i++) { 48 if (!strcasecmp(list[i], s)) return 1; 49 } 50 return 0; 51 } 52 53 static void now_plus_ms(struct timespec *out, int ms) { 54 clock_gettime(CLOCK_MONOTONIC, out); 55 out->tv_sec += ms / 1000; 56 out->tv_nsec += (long)(ms % 1000) * 1000000L; 57 if (out->tv_nsec >= 1000000000L) { out->tv_sec++; out->tv_nsec -= 1000000000L; } 58 } 59 60 static int deadline_passed(const struct timespec *t) { 61 struct timespec now; 62 clock_gettime(CLOCK_MONOTONIC, &now); 63 if (now.tv_sec != t->tv_sec) return now.tv_sec > t->tv_sec; 64 return now.tv_nsec >= t->tv_nsec; 65 } 66 67 static int set_timeouts(int fd) { 68 struct timeval tv = { .tv_sec = PLUGIN_TIMEOUT_S, .tv_usec = 0 }; 69 if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) != 0) return -1; 70 if (setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)) != 0) return -1; 71 return 0; 72 } 73 74 // socketpair, not pipes: SO_RCVTIMEO works on it, so a wedged plugin cannot 75 // block the daemon forever. 76 static int spawn_child(struct dp_plugin *p) { 77 int sv[2]; 78 if (socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0, sv) != 0) { 79 log_error("plugin %s: socketpair: %s", p->target, strerror(errno)); 80 return -1; 81 } 82 83 pid_t pid = fork(); 84 if (pid < 0) { 85 log_error("plugin %s: fork: %s", p->target, strerror(errno)); 86 close(sv[0]); 87 close(sv[1]); 88 return -1; 89 } 90 91 if (pid == 0) { 92 // stderr is inherited, so plugin diagnostics reach the daemon log. 93 dup2(sv[1], STDIN_FILENO); 94 dup2(sv[1], STDOUT_FILENO); 95 if (sv[1] > STDERR_FILENO) close(sv[1]); 96 close(sv[0]); 97 signal(SIGPIPE, SIG_DFL); 98 execl(p->target, p->target, (char *)NULL); 99 fprintf(stderr, "plugin %s: exec failed: %s\n", p->target, strerror(errno)); 100 _exit(127); 101 } 102 103 close(sv[1]); 104 if (set_timeouts(sv[0]) != 0) { 105 log_error("plugin %s: setsockopt: %s", p->target, strerror(errno)); 106 close(sv[0]); 107 return -1; 108 } 109 p->fd = sv[0]; 110 p->pid = pid; 111 return 0; 112 } 113 114 // tcp://[user:pass@]host[:port] 115 static int parse_tcp(const char *url, char **host, char **port, char **user, char **pass) { 116 const char *rest = url + 6; // skip tcp:// 117 const char *at = strrchr(rest, '@'); 118 *user = *pass = NULL; 119 120 if (at) { 121 size_t clen = (size_t)(at - rest); 122 char *cred = strndup(rest, clen); 123 if (!cred) return -1; 124 char *colon = strchr(cred, ':'); 125 if (colon) { 126 *colon = '\0'; 127 *user = strdup(cred); 128 *pass = strdup(colon + 1); 129 } else { 130 *user = strdup(cred); 131 } 132 free(cred); 133 rest = at + 1; 134 } 135 136 const char *colon = strrchr(rest, ':'); 137 if (colon) { 138 *host = strndup(rest, (size_t)(colon - rest)); 139 *port = strdup(colon + 1); 140 } else { 141 *host = strdup(rest); 142 *port = strdup("6789"); 143 } 144 return (*host && *port) ? 0 : -1; 145 } 146 147 static int connect_tcp(struct dp_plugin *p) { 148 char *host = NULL, *port = NULL; 149 free(p->user); free(p->pass); 150 p->user = p->pass = NULL; 151 152 if (parse_tcp(p->target, &host, &port, &p->user, &p->pass) != 0) { 153 log_error("plugin %s: malformed address", p->target); 154 free(host); free(port); 155 return -1; 156 } 157 158 struct addrinfo hints = { .ai_family = AF_UNSPEC, .ai_socktype = SOCK_STREAM }; 159 struct addrinfo *res = NULL; 160 int gai = getaddrinfo(host, port, &hints, &res); 161 free(host); free(port); 162 if (gai != 0) { 163 log_error("plugin %s: resolve: %s", p->target, gai_strerror(gai)); 164 return -1; 165 } 166 167 int fd = -1; 168 for (struct addrinfo *ai = res; ai; ai = ai->ai_next) { 169 fd = socket(ai->ai_family, ai->ai_socktype | SOCK_CLOEXEC, ai->ai_protocol); 170 if (fd < 0) continue; 171 if (connect(fd, ai->ai_addr, ai->ai_addrlen) == 0) break; 172 close(fd); 173 fd = -1; 174 } 175 freeaddrinfo(res); 176 177 if (fd < 0) { 178 log_error("plugin %s: connect failed: %s", p->target, strerror(errno)); 179 return -1; 180 } 181 if (set_timeouts(fd) != 0) { 182 close(fd); 183 return -1; 184 } 185 p->fd = fd; 186 p->pid = 0; 187 return 0; 188 } 189 190 static int write_all(int fd, const char *buf, size_t len) { 191 while (len) { 192 ssize_t n = write(fd, buf, len); 193 if (n <= 0) { 194 if (n < 0 && errno == EINTR) continue; 195 return -1; 196 } 197 buf += n; 198 len -= (size_t)n; 199 } 200 return 0; 201 } 202 203 resp_object *dp_plugin_call(struct dp_plugin *p, const char **argv, int argc) { 204 if (!p->up || p->fd < 0) return NULL; 205 206 resp_object *req = resp_array_init(); 207 if (!req) return NULL; 208 for (int i = 0; i < argc; i++) resp_array_append_bulk(req, argv[i]); 209 210 char *buf = NULL; 211 size_t len = 0; 212 if (resp_serialize(req, &buf, &len) != 0) { 213 resp_free(req); 214 return NULL; 215 } 216 resp_free(req); 217 218 int rc = write_all(p->fd, buf, len); 219 free(buf); 220 if (rc != 0) { 221 dp_plugin_down(p, "write failed"); 222 return NULL; 223 } 224 225 // resp_read returns NULL for timeout, EOF and parse errors alike; all three 226 // mean the plugin is unusable. 227 resp_object *reply = resp_read(p->fd); 228 if (!reply) { 229 dp_plugin_down(p, "no reply (timeout, EOF or malformed)"); 230 return NULL; 231 } 232 return reply; 233 } 234 235 static int reply_is_error(const resp_object *o) { 236 return o && o->type == RESPT_ERROR; 237 } 238 239 static void load_commands(struct dp_plugin *p) { 240 strlist_free(&p->commands, &p->commands_n); 241 const char *argv[] = { "COMMAND" }; 242 resp_object *r = dp_plugin_call(p, argv, 1); 243 if (!r) return; 244 245 if (r->type == RESPT_ARRAY) { 246 for (size_t i = 0; i < r->u.arr.n; i++) { 247 resp_object *e = &r->u.arr.elem[i]; 248 // Accept a flat list of names or redis' array-per-command form. 249 const char *name = NULL; 250 if (e->type == RESPT_BULK || e->type == RESPT_SIMPLE) { 251 name = e->u.s; 252 } else if (e->type == RESPT_ARRAY && e->u.arr.n > 0) { 253 resp_object *f = &e->u.arr.elem[0]; 254 if (f->type == RESPT_BULK || f->type == RESPT_SIMPLE) name = f->u.s; 255 } 256 if (name) strlist_add(&p->commands, &p->commands_n, name); 257 } 258 } 259 resp_free(r); 260 } 261 262 static void load_config_keys(struct dp_plugin *p) { 263 strlist_free(&p->config_keys, &p->config_keys_n); 264 if (!dp_plugin_supports(p, "CONFIG")) return; 265 266 const char *argv[] = { "CONFIG", "LIST" }; 267 resp_object *r = dp_plugin_call(p, argv, 2); 268 if (!r) return; 269 270 if (r->type == RESPT_ARRAY) { 271 for (size_t i = 0; i < r->u.arr.n; i++) { 272 resp_object *e = &r->u.arr.elem[i]; 273 if (e->type == RESPT_BULK || e->type == RESPT_SIMPLE) { 274 strlist_add(&p->config_keys, &p->config_keys_n, e->u.s); 275 } 276 } 277 } 278 resp_free(r); 279 } 280 281 // redis INFO format; hardware_detected is the only field we interpret. 282 static void load_info(struct dp_plugin *p) { 283 p->hardware = -1; 284 const char *argv[] = { "INFO" }; 285 resp_object *r = dp_plugin_call(p, argv, 1); 286 if (!r) return; 287 288 if ((r->type == RESPT_BULK || r->type == RESPT_SIMPLE) && r->u.s) { 289 const char *k = strstr(r->u.s, "hardware_detected:"); 290 if (k) p->hardware = atoi(k + strlen("hardware_detected:")) ? 1 : 0; 291 } 292 resp_free(r); 293 } 294 295 static int authenticate(struct dp_plugin *p) { 296 if (!p->pass && !p->user) return 0; 297 const char *argv[3]; 298 int argc = 0; 299 argv[argc++] = "AUTH"; 300 if (p->user && p->pass) { 301 argv[argc++] = p->user; 302 argv[argc++] = p->pass; 303 } else { 304 argv[argc++] = p->user ? p->user : p->pass; 305 } 306 resp_object *r = dp_plugin_call(p, argv, argc); 307 if (!r) return -1; 308 int bad = reply_is_error(r); 309 if (bad) log_error("plugin %s: AUTH rejected: %s", p->target, r->u.s ? r->u.s : ""); 310 resp_free(r); 311 return bad ? -1 : 0; 312 } 313 314 void dp_plugin_down(struct dp_plugin *p, const char *why) { 315 if (p->fd >= 0) { close(p->fd); p->fd = -1; } 316 if (p->pid > 0) { 317 kill(p->pid, SIGTERM); 318 waitpid(p->pid, NULL, 0); 319 p->pid = 0; 320 } 321 if (p->up) log_error("plugin %s: down (%s)", p->target, why ? why : "unknown"); 322 p->up = 0; 323 324 p->backoff_ms = p->backoff_ms ? p->backoff_ms * 2 : BACKOFF_MIN_MS; 325 if (p->backoff_ms > BACKOFF_MAX_MS) p->backoff_ms = BACKOFF_MAX_MS; 326 now_plus_ms(&p->retry_at, p->backoff_ms); 327 } 328 329 int dp_plugin_up(struct dp_plugin *p) { 330 if (p->up) return 0; 331 if (p->backoff_ms && !deadline_passed(&p->retry_at)) return -1; 332 333 int rc = !strncmp(p->target, "tcp://", 6) ? connect_tcp(p) : spawn_child(p); 334 if (rc != 0) { 335 dp_plugin_down(p, "start failed"); 336 return -1; 337 } 338 339 // Provisionally up so dp_plugin_call() will talk to it during handshake. 340 p->up = 1; 341 342 if (authenticate(p) != 0) { 343 dp_plugin_down(p, "authentication failed"); 344 return -1; 345 } 346 347 load_commands(p); 348 if (!p->up) return -1; 349 load_info(p); 350 if (!p->up) return -1; 351 load_config_keys(p); 352 if (!p->up) return -1; 353 354 if (p->hardware == 0) { 355 log_error("plugin %s: reports no supported hardware, not using it", p->target); 356 dp_plugin_down(p, "no hardware"); 357 return -1; 358 } 359 360 p->backoff_ms = 0; 361 log_info("plugin %s: up, %zu commands", p->target, p->commands_n); 362 return 0; 363 } 364 365 int dp_plugin_supports(const struct dp_plugin *p, const char *verb) { 366 return p->up && strlist_has(p->commands, p->commands_n, verb); 367 } 368 369 int dp_plugin_config_supports(const struct dp_plugin *p, const char *key) { 370 return strlist_has(p->config_keys, p->config_keys_n, key); 371 } 372 373 struct dp_plugin *dp_plugin_list(void) { 374 return plugins; 375 } 376 377 static struct dp_plugin *plugin_entry(const char *target) { 378 for (struct dp_plugin *p = plugins; p; p = p->next) { 379 if (!strcmp(p->target, target)) return p; 380 } 381 struct dp_plugin *p = calloc(1, sizeof(*p)); 382 if (!p) return NULL; 383 p->target = strdup(target); 384 if (!p->target) { free(p); return NULL; } 385 p->fd = -1; 386 struct dp_plugin **tail = &plugins; 387 while (*tail) tail = &(*tail)->next; 388 *tail = p; 389 return p; 390 } 391 392 struct dp_plugin *dp_plugin_for(const char *target) { 393 struct dp_plugin *p = plugin_entry(target); 394 if (!p) return NULL; 395 dp_plugin_up(p); 396 return p; 397 } 398 399 int dp_plugin_config_set(struct dp_plugin *p, const char *key, char **argv, size_t argc) { 400 const char *a[MAX_CONFIG_ARGV]; 401 size_t n = 0; 402 a[n++] = "CONFIG"; 403 a[n++] = "SET"; 404 a[n++] = key; 405 for (size_t i = 0; i < argc && n < MAX_CONFIG_ARGV; i++) a[n++] = argv[i]; 406 407 resp_object *r = dp_plugin_call(p, a, (int)n); 408 if (!r) return -1; 409 int bad = reply_is_error(r); 410 if (bad) { 411 log_error("plugin %s: CONFIG SET %s: %s", p->target, key, r->u.s ? r->u.s : "error"); 412 } 413 resp_free(r); 414 return bad ? -1 : 0; 415 } 416 417 int dp_start(void) { 418 const struct linkd_daemon_cfg *cfg = daemon_cfg(); 419 420 for (struct linkd_plugin_cfg *c = cfg->plugins; c; c = c->next) { 421 struct dp_plugin *p = plugin_entry(c->target); 422 if (!p) return DP_RET_ERROR; 423 p->optional = c->optional; 424 // A plugin that cannot start yet is not fatal; the backoff retries it. 425 dp_plugin_up(p); 426 } 427 return DP_RET_OK; 428 } 429 430 void dp_stop(void) { 431 struct dp_plugin *p = plugins; 432 while (p) { 433 struct dp_plugin *next = p->next; 434 if (p->fd >= 0) close(p->fd); 435 if (p->pid > 0) { 436 kill(p->pid, SIGTERM); 437 waitpid(p->pid, NULL, 0); 438 } 439 strlist_free(&p->commands, &p->commands_n); 440 strlist_free(&p->config_keys, &p->config_keys_n); 441 free(p->target); 442 free(p->user); 443 free(p->pass); 444 free(p); 445 p = next; 446 } 447 plugins = NULL; 448 } 449 450 int dp_have_plugins(void) { 451 return plugins != NULL; 452 }