1 /*
2 * Stream processing offload engine management.
3 *
4 * Copyright 2016 HAProxy Technologies, Christopher Faulet <cfaulet@haproxy.com>
5 *
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU General Public License
8 * as published by the Free Software Foundation; either version
9 * 2 of the License, or (at your option) any later version.
10 *
11 */
12 #include <ctype.h>
13 #include <errno.h>
14
15 #include <common/cfgparse.h>
16 #include <common/compat.h>
17 #include <common/config.h>
18 #include <common/debug.h>
19 #include <common/hathreads.h>
20 #include <common/initcall.h>
21 #include <common/memory.h>
22 #include <common/time.h>
23
24 #include <types/arg.h>
25 #include <types/global.h>
26 #include <types/spoe.h>
27
28 #include <proto/acl.h>
29 #include <proto/action.h>
30 #include <proto/arg.h>
31 #include <proto/backend.h>
32 #include <proto/filters.h>
33 #include <proto/freq_ctr.h>
34 #include <proto/frontend.h>
35 #include <proto/http_rules.h>
36 #include <proto/log.h>
37 #include <proto/http_ana.h>
38 #include <proto/proxy.h>
39 #include <proto/sample.h>
40 #include <proto/session.h>
41 #include <proto/signal.h>
42 #include <proto/spoe.h>
43 #include <proto/stream.h>
44 #include <proto/stream_interface.h>
45 #include <proto/task.h>
46 #include <proto/tcp_rules.h>
47 #include <proto/vars.h>
48
49 #if defined(DEBUG_SPOE) || defined(DEBUG_FULL)
50 #define SPOE_PRINTF(x...) fprintf(x)
51 #define SPOE_DEBUG_STMT(statement) statement
52 #else
53 #define SPOE_PRINTF(x...)
54 #define SPOE_DEBUG_STMT(statement)
55 #endif
56
57 /* Reserved 4 bytes to the frame size. So a frame and its size can be written
58 * together in a buffer */
59 #define MAX_FRAME_SIZE global.tune.bufsize - 4
60
61 /* The minimum size for a frame */
62 #define MIN_FRAME_SIZE 256
63
64 /* Reserved for the metadata and the frame type.
65 * So <MAX_FRAME_SIZE> - <FRAME_HDR_SIZE> is the maximum payload size */
66 #define FRAME_HDR_SIZE 32
67
68 /* Helper to get SPOE ctx inside an appctx */
69 #define SPOE_APPCTX(appctx) ((struct spoe_appctx *)((appctx)->ctx.spoe.ptr))
70
71 /* SPOE filter id. Used to identify SPOE filters */
72 const char *spoe_filter_id = "SPOE filter";
73
74 /* Set if the handle on SIGUSR1 is registered */
75 static int sighandler_registered = 0;
76
77 /* proxy used during the parsing */
78 struct proxy *curproxy = NULL;
79
80 /* The name of the SPOE engine, used during the parsing */
81 char *curengine = NULL;
82
83 /* SPOE agent used during the parsing */
84 /* SPOE agent/group/message used during the parsing */
85 struct spoe_agent *curagent = NULL;
86 struct spoe_group *curgrp = NULL;
87 struct spoe_message *curmsg = NULL;
88
89 /* list of SPOE messages and placeholders used during the parsing */
90 struct list curmsgs;
91 struct list curgrps;
92 struct list curmphs;
93 struct list curgphs;
94 struct list curvars;
95
96 /* list of log servers used during the parsing */
97 struct list curlogsrvs;
98
99 /* agent's proxy flags (PR_O_* and PR_O2_*) used during parsing */
100 int curpxopts;
101 int curpxopts2;
102
103 /* Pools used to allocate SPOE structs */
104 DECLARE_STATIC_POOL(pool_head_spoe_ctx, "spoe_ctx", sizeof(struct spoe_context));
105 DECLARE_STATIC_POOL(pool_head_spoe_appctx, "spoe_appctx", sizeof(struct spoe_appctx));
106
107 struct flt_ops spoe_ops;
108
109 static int spoe_queue_context(struct spoe_context *ctx);
110 static int spoe_acquire_buffer(struct buffer *buf, struct buffer_wait *buffer_wait);
111 static void spoe_release_buffer(struct buffer *buf, struct buffer_wait *buffer_wait);
112
113 /********************************************************************
114 * helper functions/globals
115 ********************************************************************/
116 static void
spoe_release_placeholder(struct spoe_placeholder * ph)117 spoe_release_placeholder(struct spoe_placeholder *ph)
118 {
119 if (!ph)
120 return;
121 free(ph->id);
122 free(ph);
123 }
124
125 static void
spoe_release_message(struct spoe_message * msg)126 spoe_release_message(struct spoe_message *msg)
127 {
128 struct spoe_arg *arg, *argback;
129 struct acl *acl, *aclback;
130
131 if (!msg)
132 return;
133 free(msg->id);
134 free(msg->conf.file);
135 list_for_each_entry_safe(arg, argback, &msg->args, list) {
136 release_sample_expr(arg->expr);
137 free(arg->name);
138 LIST_DEL(&arg->list);
139 free(arg);
140 }
141 list_for_each_entry_safe(acl, aclback, &msg->acls, list) {
142 LIST_DEL(&acl->list);
143 prune_acl(acl);
144 free(acl);
145 }
146 if (msg->cond) {
147 prune_acl_cond(msg->cond);
148 free(msg->cond);
149 }
150 free(msg);
151 }
152
153 static void
spoe_release_group(struct spoe_group * grp)154 spoe_release_group(struct spoe_group *grp)
155 {
156 if (!grp)
157 return;
158 free(grp->id);
159 free(grp->conf.file);
160 free(grp);
161 }
162
163 static void
spoe_release_agent(struct spoe_agent * agent)164 spoe_release_agent(struct spoe_agent *agent)
165 {
166 struct spoe_message *msg, *msgback;
167 struct spoe_group *grp, *grpback;
168 int i;
169
170 if (!agent)
171 return;
172 free(agent->id);
173 free(agent->conf.file);
174 free(agent->var_pfx);
175 free(agent->var_on_error);
176 free(agent->var_t_process);
177 free(agent->var_t_total);
178 list_for_each_entry_safe(msg, msgback, &agent->messages, list) {
179 LIST_DEL(&msg->list);
180 spoe_release_message(msg);
181 }
182 list_for_each_entry_safe(grp, grpback, &agent->groups, list) {
183 LIST_DEL(&grp->list);
184 spoe_release_group(grp);
185 }
186 if (agent->rt) {
187 for (i = 0; i < global.nbthread; ++i) {
188 free(agent->rt[i].engine_id);
189 HA_SPIN_DESTROY(&agent->rt[i].lock);
190 }
191 }
192 free(agent->rt);
193 free(agent);
194 }
195
196 static const char *spoe_frm_err_reasons[SPOE_FRM_ERRS] = {
197 [SPOE_FRM_ERR_NONE] = "normal",
198 [SPOE_FRM_ERR_IO] = "I/O error",
199 [SPOE_FRM_ERR_TOUT] = "a timeout occurred",
200 [SPOE_FRM_ERR_TOO_BIG] = "frame is too big",
201 [SPOE_FRM_ERR_INVALID] = "invalid frame received",
202 [SPOE_FRM_ERR_NO_VSN] = "version value not found",
203 [SPOE_FRM_ERR_NO_FRAME_SIZE] = "max-frame-size value not found",
204 [SPOE_FRM_ERR_NO_CAP] = "capabilities value not found",
205 [SPOE_FRM_ERR_BAD_VSN] = "unsupported version",
206 [SPOE_FRM_ERR_BAD_FRAME_SIZE] = "max-frame-size too big or too small",
207 [SPOE_FRM_ERR_FRAG_NOT_SUPPORTED] = "fragmentation not supported",
208 [SPOE_FRM_ERR_INTERLACED_FRAMES] = "invalid interlaced frames",
209 [SPOE_FRM_ERR_FRAMEID_NOTFOUND] = "frame-id not found",
210 [SPOE_FRM_ERR_RES] = "resource allocation error",
211 [SPOE_FRM_ERR_UNKNOWN] = "an unknown error occurred",
212 };
213
214 static const char *spoe_event_str[SPOE_EV_EVENTS] = {
215 [SPOE_EV_ON_CLIENT_SESS] = "on-client-session",
216 [SPOE_EV_ON_TCP_REQ_FE] = "on-frontend-tcp-request",
217 [SPOE_EV_ON_TCP_REQ_BE] = "on-backend-tcp-request",
218 [SPOE_EV_ON_HTTP_REQ_FE] = "on-frontend-http-request",
219 [SPOE_EV_ON_HTTP_REQ_BE] = "on-backend-http-request",
220
221 [SPOE_EV_ON_SERVER_SESS] = "on-server-session",
222 [SPOE_EV_ON_TCP_RSP] = "on-tcp-response",
223 [SPOE_EV_ON_HTTP_RSP] = "on-http-response",
224 };
225
226
227 #if defined(DEBUG_SPOE) || defined(DEBUG_FULL)
228
229 static const char *spoe_ctx_state_str[SPOE_CTX_ST_ERROR+1] = {
230 [SPOE_CTX_ST_NONE] = "NONE",
231 [SPOE_CTX_ST_READY] = "READY",
232 [SPOE_CTX_ST_ENCODING_MSGS] = "ENCODING_MSGS",
233 [SPOE_CTX_ST_SENDING_MSGS] = "SENDING_MSGS",
234 [SPOE_CTX_ST_WAITING_ACK] = "WAITING_ACK",
235 [SPOE_CTX_ST_DONE] = "DONE",
236 [SPOE_CTX_ST_ERROR] = "ERROR",
237 };
238
239 static const char *spoe_appctx_state_str[SPOE_APPCTX_ST_END+1] = {
240 [SPOE_APPCTX_ST_CONNECT] = "CONNECT",
241 [SPOE_APPCTX_ST_CONNECTING] = "CONNECTING",
242 [SPOE_APPCTX_ST_IDLE] = "IDLE",
243 [SPOE_APPCTX_ST_PROCESSING] = "PROCESSING",
244 [SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY] = "SENDING_FRAG_NOTIFY",
245 [SPOE_APPCTX_ST_WAITING_SYNC_ACK] = "WAITING_SYNC_ACK",
246 [SPOE_APPCTX_ST_DISCONNECT] = "DISCONNECT",
247 [SPOE_APPCTX_ST_DISCONNECTING] = "DISCONNECTING",
248 [SPOE_APPCTX_ST_EXIT] = "EXIT",
249 [SPOE_APPCTX_ST_END] = "END",
250 };
251
252 #endif
253
254 /* Used to generates a unique id for an engine. On success, it returns a
255 * allocated string. So it is the caller's reponsibility to release it. If the
256 * allocation failed, it returns NULL. */
257 static char *
generate_pseudo_uuid()258 generate_pseudo_uuid()
259 {
260 char *uuid;
261 uint32_t rnd[4] = { 0, 0, 0, 0 };
262 uint64_t last = 0;
263 int byte = 0;
264 uint8_t bits = 0;
265 unsigned int rand_max_bits = my_flsl(RAND_MAX);
266
267 if ((uuid = calloc(1, 37)) == NULL)
268 return NULL;
269
270 while (byte < 4) {
271 while (bits < 32) {
272 last |= (uint64_t)ha_random() << bits;
273 bits += rand_max_bits;
274 }
275 rnd[byte++] = last;
276 last >>= 32u;
277 bits -= 32;
278 }
279 snprintf(uuid, 37, "%8.8x-%4.4x-%4.4x-%4.4x-%12.12llx",
280 rnd[0],
281 rnd[1] & 0xFFFF,
282 ((rnd[1] >> 16u) & 0xFFF) | 0x4000, // highest 4 bits indicate the uuid version
283 (rnd[2] & 0x3FFF) | 0x8000, // the highest 2 bits indicate the UUID variant (10),
284 (long long)((rnd[2] >> 14u) | ((uint64_t) rnd[3] << 18u)) & 0xFFFFFFFFFFFFull
285 );
286 return uuid;
287 }
288
289
290 static inline void
spoe_update_stat_time(struct timeval * tv,long * t)291 spoe_update_stat_time(struct timeval *tv, long *t)
292 {
293 if (*t == -1)
294 *t = tv_ms_elapsed(tv, &now);
295 else
296 *t += tv_ms_elapsed(tv, &now);
297 tv_zero(tv);
298 }
299
300 /********************************************************************
301 * Functions that encode/decode SPOE frames
302 ********************************************************************/
303 /* Helper to get static string length, excluding the terminating null byte */
304 #define SLEN(str) (sizeof(str)-1)
305
306 /* Predefined key used in HELLO/DISCONNECT frames */
307 #define SUPPORTED_VERSIONS_KEY "supported-versions"
308 #define VERSION_KEY "version"
309 #define MAX_FRAME_SIZE_KEY "max-frame-size"
310 #define CAPABILITIES_KEY "capabilities"
311 #define ENGINE_ID_KEY "engine-id"
312 #define HEALTHCHECK_KEY "healthcheck"
313 #define STATUS_CODE_KEY "status-code"
314 #define MSG_KEY "message"
315
316 struct spoe_version {
317 char *str;
318 int min;
319 int max;
320 };
321
322 /* All supported versions */
323 static struct spoe_version supported_versions[] = {
324 /* 1.0 is now unsupported because of a bug about frame's flags*/
325 {"2.0", 2000, 2000},
326 {NULL, 0, 0}
327 };
328
329 /* Comma-separated list of supported versions */
330 #define SUPPORTED_VERSIONS_VAL "2.0"
331
332 /* Convert a string to a SPOE version value. The string must follow the format
333 * "MAJOR.MINOR". It will be concerted into the integer (1000 * MAJOR + MINOR).
334 * If an error occurred, -1 is returned. */
335 static int
spoe_str_to_vsn(const char * str,size_t len)336 spoe_str_to_vsn(const char *str, size_t len)
337 {
338 const char *p, *end;
339 int maj, min, vsn;
340
341 p = str;
342 end = str+len;
343 maj = min = 0;
344 vsn = -1;
345
346 /* skip leading spaces */
347 while (p < end && isspace(*p))
348 p++;
349
350 /* parse Major number, until the '.' */
351 while (*p != '.') {
352 if (p >= end || *p < '0' || *p > '9')
353 goto out;
354 maj *= 10;
355 maj += (*p - '0');
356 p++;
357 }
358
359 /* check Major version */
360 if (!maj)
361 goto out;
362
363 p++; /* skip the '.' */
364 if (p >= end || *p < '0' || *p > '9') /* Minor number is missing */
365 goto out;
366
367 /* Parse Minor number */
368 while (p < end) {
369 if (*p < '0' || *p > '9')
370 break;
371 min *= 10;
372 min += (*p - '0');
373 p++;
374 }
375
376 /* check Minor number */
377 if (min > 999)
378 goto out;
379
380 /* skip trailing spaces */
381 while (p < end && isspace(*p))
382 p++;
383 if (p != end)
384 goto out;
385
386 vsn = maj * 1000 + min;
387 out:
388 return vsn;
389 }
390
391 /* Encode the HELLO frame sent by HAProxy to an agent. It returns the number of
392 * encoded bytes in the frame on success, 0 if an encoding error occurred and -1
393 * if a fatal error occurred. */
394 static int
spoe_prepare_hahello_frame(struct appctx * appctx,char * frame,size_t size)395 spoe_prepare_hahello_frame(struct appctx *appctx, char *frame, size_t size)
396 {
397 struct buffer *chk;
398 struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
399 char *p, *end;
400 unsigned int flags = SPOE_FRM_FL_FIN;
401 size_t sz;
402
403 p = frame;
404 end = frame+size;
405
406 /* Set Frame type */
407 *p++ = SPOE_FRM_T_HAPROXY_HELLO;
408
409 /* Set flags */
410 flags = htonl(flags);
411 memcpy(p, (char *)&flags, 4);
412 p += 4;
413
414 /* No stream-id and frame-id for HELLO frames */
415 *p++ = 0; *p++ = 0;
416
417 /* There are 3 mandatory items: "supported-versions", "max-frame-size"
418 * and "capabilities" */
419
420 /* "supported-versions" K/V item */
421 sz = SLEN(SUPPORTED_VERSIONS_KEY);
422 if (spoe_encode_buffer(SUPPORTED_VERSIONS_KEY, sz, &p, end) == -1)
423 goto too_big;
424
425 *p++ = SPOE_DATA_T_STR;
426 sz = SLEN(SUPPORTED_VERSIONS_VAL);
427 if (spoe_encode_buffer(SUPPORTED_VERSIONS_VAL, sz, &p, end) == -1)
428 goto too_big;
429
430 /* "max-fram-size" K/V item */
431 sz = SLEN(MAX_FRAME_SIZE_KEY);
432 if (spoe_encode_buffer(MAX_FRAME_SIZE_KEY, sz, &p, end) == -1)
433 goto too_big;
434
435 *p++ = SPOE_DATA_T_UINT32;
436 if (encode_varint(SPOE_APPCTX(appctx)->max_frame_size, &p, end) == -1)
437 goto too_big;
438
439 /* "capabilities" K/V item */
440 sz = SLEN(CAPABILITIES_KEY);
441 if (spoe_encode_buffer(CAPABILITIES_KEY, sz, &p, end) == -1)
442 goto too_big;
443
444 *p++ = SPOE_DATA_T_STR;
445 chk = get_trash_chunk();
446 if (agent != NULL && (agent->flags & SPOE_FL_PIPELINING)) {
447 memcpy(chk->area, "pipelining", 10);
448 chk->data += 10;
449 }
450 if (agent != NULL && (agent->flags & SPOE_FL_ASYNC)) {
451 if (chk->data) chk->area[chk->data++] = ',';
452 memcpy(chk->area+chk->data, "async", 5);
453 chk->data += 5;
454 }
455 if (agent != NULL && (agent->flags & SPOE_FL_RCV_FRAGMENTATION)) {
456 if (chk->data) chk->area[chk->data++] = ',';
457 memcpy(chk->area+chk->data, "fragmentation", 13);
458 chk->data += 13;
459 }
460 if (spoe_encode_buffer(chk->area, chk->data, &p, end) == -1)
461 goto too_big;
462
463 /* (optionnal) "engine-id" K/V item, if present */
464 if (agent != NULL && agent->rt[tid].engine_id != NULL) {
465 sz = SLEN(ENGINE_ID_KEY);
466 if (spoe_encode_buffer(ENGINE_ID_KEY, sz, &p, end) == -1)
467 goto too_big;
468
469 *p++ = SPOE_DATA_T_STR;
470 sz = strlen(agent->rt[tid].engine_id);
471 if (spoe_encode_buffer(agent->rt[tid].engine_id, sz, &p, end) == -1)
472 goto too_big;
473 }
474
475 return (p - frame);
476
477 too_big:
478 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
479 return 0;
480 }
481
482 /* Encode DISCONNECT frame sent by HAProxy to an agent. It returns the number of
483 * encoded bytes in the frame on success, 0 if an encoding error occurred and -1
484 * if a fatal error occurred. */
485 static int
spoe_prepare_hadiscon_frame(struct appctx * appctx,char * frame,size_t size)486 spoe_prepare_hadiscon_frame(struct appctx *appctx, char *frame, size_t size)
487 {
488 const char *reason;
489 char *p, *end;
490 unsigned int flags = SPOE_FRM_FL_FIN;
491 size_t sz;
492
493 p = frame;
494 end = frame+size;
495
496 /* Set Frame type */
497 *p++ = SPOE_FRM_T_HAPROXY_DISCON;
498
499 /* Set flags */
500 flags = htonl(flags);
501 memcpy(p, (char *)&flags, 4);
502 p += 4;
503
504 /* No stream-id and frame-id for DISCONNECT frames */
505 *p++ = 0; *p++ = 0;
506
507 if (SPOE_APPCTX(appctx)->status_code >= SPOE_FRM_ERRS)
508 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_UNKNOWN;
509
510 /* There are 2 mandatory items: "status-code" and "message" */
511
512 /* "status-code" K/V item */
513 sz = SLEN(STATUS_CODE_KEY);
514 if (spoe_encode_buffer(STATUS_CODE_KEY, sz, &p, end) == -1)
515 goto too_big;
516
517 *p++ = SPOE_DATA_T_UINT32;
518 if (encode_varint(SPOE_APPCTX(appctx)->status_code, &p, end) == -1)
519 goto too_big;
520
521 /* "message" K/V item */
522 sz = SLEN(MSG_KEY);
523 if (spoe_encode_buffer(MSG_KEY, sz, &p, end) == -1)
524 goto too_big;
525
526 /*Get the message corresponding to the status code */
527 reason = spoe_frm_err_reasons[SPOE_APPCTX(appctx)->status_code];
528
529 *p++ = SPOE_DATA_T_STR;
530 sz = strlen(reason);
531 if (spoe_encode_buffer(reason, sz, &p, end) == -1)
532 goto too_big;
533
534 return (p - frame);
535
536 too_big:
537 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
538 return 0;
539 }
540
541 /* Encode the NOTIFY frame sent by HAProxy to an agent. It returns the number of
542 * encoded bytes in the frame on success, 0 if an encoding error occurred and -1
543 * if a fatal error occurred. */
544 static int
spoe_prepare_hanotify_frame(struct appctx * appctx,struct spoe_context * ctx,char * frame,size_t size)545 spoe_prepare_hanotify_frame(struct appctx *appctx, struct spoe_context *ctx,
546 char *frame, size_t size)
547 {
548 char *p, *end;
549 unsigned int stream_id, frame_id;
550 unsigned int flags = SPOE_FRM_FL_FIN;
551 size_t sz;
552
553 p = frame;
554 end = frame+size;
555
556 stream_id = ctx->stream_id;
557 frame_id = ctx->frame_id;
558
559 if (ctx->flags & SPOE_CTX_FL_FRAGMENTED) {
560 /* The fragmentation is not supported by the applet */
561 if (!(SPOE_APPCTX(appctx)->flags & SPOE_APPCTX_FL_FRAGMENTATION)) {
562 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
563 return -1;
564 }
565 flags = ctx->frag_ctx.flags;
566 }
567
568 /* Set Frame type */
569 *p++ = SPOE_FRM_T_HAPROXY_NOTIFY;
570
571 /* Set flags */
572 flags = htonl(flags);
573 memcpy(p, (char *)&flags, 4);
574 p += 4;
575
576 /* Set stream-id and frame-id */
577 if (encode_varint(stream_id, &p, end) == -1)
578 goto too_big;
579 if (encode_varint(frame_id, &p, end) == -1)
580 goto too_big;
581
582 /* Copy encoded messages, if possible */
583 sz = b_data(&ctx->buffer);
584 if (p + sz >= end)
585 goto too_big;
586 memcpy(p, b_head(&ctx->buffer), sz);
587 p += sz;
588
589 return (p - frame);
590
591 too_big:
592 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
593 return 0;
594 }
595
596 /* Encode next part of a fragmented frame sent by HAProxy to an agent. It
597 * returns the number of encoded bytes in the frame on success, 0 if an encoding
598 * error occurred and -1 if a fatal error occurred. */
599 static int
spoe_prepare_hafrag_frame(struct appctx * appctx,struct spoe_context * ctx,char * frame,size_t size)600 spoe_prepare_hafrag_frame(struct appctx *appctx, struct spoe_context *ctx,
601 char *frame, size_t size)
602 {
603 char *p, *end;
604 unsigned int stream_id, frame_id;
605 unsigned int flags;
606 size_t sz;
607
608 p = frame;
609 end = frame+size;
610
611 /* <ctx> is null when the stream has aborted the processing of a
612 * fragmented frame. In this case, we must notify the corresponding
613 * agent using ids stored in <frag_ctx>. */
614 if (ctx == NULL) {
615 flags = (SPOE_FRM_FL_FIN|SPOE_FRM_FL_ABRT);
616 stream_id = SPOE_APPCTX(appctx)->frag_ctx.cursid;
617 frame_id = SPOE_APPCTX(appctx)->frag_ctx.curfid;
618 }
619 else {
620 flags = ctx->frag_ctx.flags;
621 stream_id = ctx->stream_id;
622 frame_id = ctx->frame_id;
623 }
624
625 /* Set Frame type */
626 *p++ = SPOE_FRM_T_UNSET;
627
628 /* Set flags */
629 flags = htonl(flags);
630 memcpy(p, (char *)&flags, 4);
631 p += 4;
632
633 /* Set stream-id and frame-id */
634 if (encode_varint(stream_id, &p, end) == -1)
635 goto too_big;
636 if (encode_varint(frame_id, &p, end) == -1)
637 goto too_big;
638
639 if (ctx == NULL)
640 goto end;
641
642 /* Copy encoded messages, if possible */
643 sz = b_data(&ctx->buffer);
644 if (p + sz >= end)
645 goto too_big;
646 memcpy(p, b_head(&ctx->buffer), sz);
647 p += sz;
648
649 end:
650 return (p - frame);
651
652 too_big:
653 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
654 return 0;
655 }
656
657 /* Decode and process the HELLO frame sent by an agent. It returns the number of
658 * read bytes on success, 0 if a decoding error occurred, and -1 if a fatal
659 * error occurred. */
660 static int
spoe_handle_agenthello_frame(struct appctx * appctx,char * frame,size_t size)661 spoe_handle_agenthello_frame(struct appctx *appctx, char *frame, size_t size)
662 {
663 struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
664 char *p, *end;
665 int vsn, max_frame_size;
666 unsigned int flags;
667
668 p = frame;
669 end = frame + size;
670
671 /* Check frame type */
672 if (*p++ != SPOE_FRM_T_AGENT_HELLO) {
673 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
674 return 0;
675 }
676
677 if (size < 7 /* TYPE + METADATA */) {
678 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
679 return 0;
680 }
681
682 /* Retrieve flags */
683 memcpy((char *)&flags, p, 4);
684 flags = ntohl(flags);
685 p += 4;
686
687 /* Fragmentation is not supported for HELLO frame */
688 if (!(flags & SPOE_FRM_FL_FIN)) {
689 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
690 return -1;
691 }
692
693 /* stream-id and frame-id must be cleared */
694 if (*p != 0 || *(p+1) != 0) {
695 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
696 return 0;
697 }
698 p += 2;
699
700 /* There are 3 mandatory items: "version", "max-frame-size" and
701 * "capabilities" */
702
703 /* Loop on K/V items */
704 vsn = max_frame_size = flags = 0;
705 while (p < end) {
706 char *str;
707 uint64_t sz;
708 int ret;
709
710 /* Decode the item key */
711 ret = spoe_decode_buffer(&p, end, &str, &sz);
712 if (ret == -1 || !sz) {
713 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
714 return 0;
715 }
716
717 /* Check "version" K/V item */
718 if (sz >= strlen(VERSION_KEY) && !memcmp(str, VERSION_KEY, strlen(VERSION_KEY))) {
719 int i, type = *p++;
720
721 /* The value must be a string */
722 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR) {
723 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
724 return 0;
725 }
726 if (spoe_decode_buffer(&p, end, &str, &sz) == -1) {
727 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
728 return 0;
729 }
730
731 vsn = spoe_str_to_vsn(str, sz);
732 if (vsn == -1) {
733 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_BAD_VSN;
734 return -1;
735 }
736 for (i = 0; supported_versions[i].str != NULL; ++i) {
737 if (vsn >= supported_versions[i].min &&
738 vsn <= supported_versions[i].max)
739 break;
740 }
741 if (supported_versions[i].str == NULL) {
742 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_BAD_VSN;
743 return -1;
744 }
745 }
746 /* Check "max-frame-size" K/V item */
747 else if (sz >= strlen(MAX_FRAME_SIZE_KEY) && !memcmp(str, MAX_FRAME_SIZE_KEY, strlen(MAX_FRAME_SIZE_KEY))) {
748 int type = *p++;
749
750 /* The value must be integer */
751 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
752 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
753 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
754 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64) {
755 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
756 return 0;
757 }
758 if (decode_varint(&p, end, &sz) == -1) {
759 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
760 return 0;
761 }
762 if (sz < MIN_FRAME_SIZE ||
763 sz > SPOE_APPCTX(appctx)->max_frame_size) {
764 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_BAD_FRAME_SIZE;
765 return -1;
766 }
767 max_frame_size = sz;
768 }
769 /* Check "capabilities" K/V item */
770 else if (sz >= strlen(CAPABILITIES_KEY) && !memcmp(str, CAPABILITIES_KEY, strlen(CAPABILITIES_KEY))) {
771 int type = *p++;
772
773 /* The value must be a string */
774 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR) {
775 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
776 return 0;
777 }
778 if (spoe_decode_buffer(&p, end, &str, &sz) == -1) {
779 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
780 return 0;
781 }
782
783 while (sz) {
784 char *delim;
785
786 /* Skip leading spaces */
787 for (; isspace(*str) && sz; str++, sz--);
788
789 if (sz >= 10 && !strncmp(str, "pipelining", 10)) {
790 str += 10; sz -= 10;
791 if (!sz || isspace(*str) || *str == ',')
792 flags |= SPOE_APPCTX_FL_PIPELINING;
793 }
794 else if (sz >= 5 && !strncmp(str, "async", 5)) {
795 str += 5; sz -= 5;
796 if (!sz || isspace(*str) || *str == ',')
797 flags |= SPOE_APPCTX_FL_ASYNC;
798 }
799 else if (sz >= 13 && !strncmp(str, "fragmentation", 13)) {
800 str += 13; sz -= 13;
801 if (!sz || isspace(*str) || *str == ',')
802 flags |= SPOE_APPCTX_FL_FRAGMENTATION;
803 }
804
805 /* Get the next comma or break */
806 if (!sz || (delim = memchr(str, ',', sz)) == NULL)
807 break;
808 delim++;
809 sz -= (delim - str);
810 str = delim;
811 }
812 }
813 else {
814 /* Silently ignore unknown item */
815 if (spoe_skip_data(&p, end) == -1) {
816 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
817 return 0;
818 }
819 }
820 }
821
822 /* Final checks */
823 if (!vsn) {
824 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NO_VSN;
825 return -1;
826 }
827 if (!max_frame_size) {
828 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NO_FRAME_SIZE;
829 return -1;
830 }
831 if (!agent)
832 flags &= ~(SPOE_APPCTX_FL_PIPELINING|SPOE_APPCTX_FL_ASYNC);
833 else {
834 if ((flags & SPOE_APPCTX_FL_PIPELINING) && !(agent->flags & SPOE_FL_PIPELINING))
835 flags &= ~SPOE_APPCTX_FL_PIPELINING;
836 if ((flags & SPOE_APPCTX_FL_ASYNC) && !(agent->flags & SPOE_FL_ASYNC))
837 flags &= ~SPOE_APPCTX_FL_ASYNC;
838 }
839
840 SPOE_APPCTX(appctx)->version = (unsigned int)vsn;
841 SPOE_APPCTX(appctx)->max_frame_size = (unsigned int)max_frame_size;
842 SPOE_APPCTX(appctx)->flags |= flags;
843
844 return (p - frame);
845 }
846
847 /* Decode DISCONNECT frame sent by an agent. It returns the number of by read
848 * bytes on success, 0 if the frame can be ignored and -1 if an error
849 * occurred. */
850 static int
spoe_handle_agentdiscon_frame(struct appctx * appctx,char * frame,size_t size)851 spoe_handle_agentdiscon_frame(struct appctx *appctx, char *frame, size_t size)
852 {
853 char *p, *end;
854 unsigned int flags;
855
856 p = frame;
857 end = frame + size;
858
859 /* Check frame type */
860 if (*p++ != SPOE_FRM_T_AGENT_DISCON) {
861 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
862 return 0;
863 }
864
865 if (size < 7 /* TYPE + METADATA */) {
866 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
867 return 0;
868 }
869
870 /* Retrieve flags */
871 memcpy((char *)&flags, p, 4);
872 flags = ntohl(flags);
873 p += 4;
874
875 /* Fragmentation is not supported for DISCONNECT frame */
876 if (!(flags & SPOE_FRM_FL_FIN)) {
877 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
878 return -1;
879 }
880
881 /* stream-id and frame-id must be cleared */
882 if (*p != 0 || *(p+1) != 0) {
883 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
884 return 0;
885 }
886 p += 2;
887
888 /* There are 2 mandatory items: "status-code" and "message" */
889
890 /* Loop on K/V items */
891 while (p < end) {
892 char *str;
893 uint64_t sz;
894 int ret;
895
896 /* Decode the item key */
897 ret = spoe_decode_buffer(&p, end, &str, &sz);
898 if (ret == -1 || !sz) {
899 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
900 return 0;
901 }
902
903 /* Check "status-code" K/V item */
904 if (sz >= strlen(STATUS_CODE_KEY) && !memcmp(str, STATUS_CODE_KEY, strlen(STATUS_CODE_KEY))) {
905 int type = *p++;
906
907 /* The value must be an integer */
908 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
909 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
910 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
911 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64) {
912 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
913 return 0;
914 }
915 if (decode_varint(&p, end, &sz) == -1) {
916 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
917 return 0;
918 }
919 SPOE_APPCTX(appctx)->status_code = sz;
920 }
921
922 /* Check "message" K/V item */
923 else if (sz >= strlen(MSG_KEY) && !memcmp(str, MSG_KEY, strlen(MSG_KEY))) {
924 int type = *p++;
925
926 /* The value must be a string */
927 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR) {
928 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
929 return 0;
930 }
931 ret = spoe_decode_buffer(&p, end, &str, &sz);
932 if (ret == -1 || sz > 255) {
933 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
934 return 0;
935 }
936 #if defined(DEBUG_SPOE) || defined(DEBUG_FULL)
937 SPOE_APPCTX(appctx)->reason = str;
938 SPOE_APPCTX(appctx)->rlen = sz;
939 #endif
940 }
941 else {
942 /* Silently ignore unknown item */
943 if (spoe_skip_data(&p, end) == -1) {
944 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
945 return 0;
946 }
947 }
948 }
949
950 return (p - frame);
951 }
952
953
954 /* Decode ACK frame sent by an agent. It returns the number of read bytes on
955 * success, 0 if the frame can be ignored and -1 if an error occurred. */
956 static int
spoe_handle_agentack_frame(struct appctx * appctx,struct spoe_context ** ctx,char * frame,size_t size)957 spoe_handle_agentack_frame(struct appctx *appctx, struct spoe_context **ctx,
958 char *frame, size_t size)
959 {
960 struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
961 char *p, *end;
962 uint64_t stream_id, frame_id;
963 int len;
964 unsigned int flags;
965
966 p = frame;
967 end = frame + size;
968 *ctx = NULL;
969
970 /* Check frame type */
971 if (*p++ != SPOE_FRM_T_AGENT_ACK) {
972 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
973 return 0;
974 }
975
976 if (size < 7 /* TYPE + METADATA */) {
977 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
978 return 0;
979 }
980
981 /* Retrieve flags */
982 memcpy((char *)&flags, p, 4);
983 flags = ntohl(flags);
984 p += 4;
985
986 /* Fragmentation is not supported for now */
987 if (!(flags & SPOE_FRM_FL_FIN)) {
988 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
989 return -1;
990 }
991
992 /* Get the stream-id and the frame-id */
993 if (decode_varint(&p, end, &stream_id) == -1) {
994 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
995 return 0;
996 }
997 if (decode_varint(&p, end, &frame_id) == -1) {
998 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
999 return 0;
1000 }
1001
1002 /* Try to find the corresponding SPOE context */
1003 if (SPOE_APPCTX(appctx)->flags & SPOE_APPCTX_FL_ASYNC) {
1004 list_for_each_entry((*ctx), &agent->rt[tid].waiting_queue, list) {
1005 if ((*ctx)->stream_id == (unsigned int)stream_id &&
1006 (*ctx)->frame_id == (unsigned int)frame_id)
1007 goto found;
1008 }
1009 }
1010 else {
1011 list_for_each_entry((*ctx), &SPOE_APPCTX(appctx)->waiting_queue, list) {
1012 if ((*ctx)->stream_id == (unsigned int)stream_id &&
1013 (*ctx)->frame_id == (unsigned int)frame_id)
1014 goto found;
1015 }
1016 }
1017
1018 if (SPOE_APPCTX(appctx)->frag_ctx.ctx &&
1019 SPOE_APPCTX(appctx)->frag_ctx.cursid == (unsigned int)stream_id &&
1020 SPOE_APPCTX(appctx)->frag_ctx.curfid == (unsigned int)frame_id) {
1021
1022 /* ABRT bit is set for an unfinished fragmented frame */
1023 if (flags & SPOE_FRM_FL_ABRT) {
1024 *ctx = SPOE_APPCTX(appctx)->frag_ctx.ctx;
1025 (*ctx)->state = SPOE_CTX_ST_ERROR;
1026 (*ctx)->status_code = SPOE_CTX_ERR_FRAG_FRAME_ABRT;
1027 /* Ignore the payload */
1028 goto end;
1029 }
1030 /* TODO: Handle more flags for fragmented frames: RESUME, FINISH... */
1031 /* For now, we ignore the ack */
1032 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
1033 return 0;
1034 }
1035
1036 /* No Stream found, ignore the frame */
1037 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1038 " - Ignore ACK frame"
1039 " - stream-id=%u - frame-id=%u\n",
1040 (int)now.tv_sec, (int)now.tv_usec, agent->id,
1041 __FUNCTION__, appctx,
1042 (unsigned int)stream_id, (unsigned int)frame_id);
1043
1044 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAMEID_NOTFOUND;
1045 if (appctx->st0 == SPOE_APPCTX_ST_WAITING_SYNC_ACK) {
1046 /* Report an error if we are waiting the ack for another frame,
1047 * but not if there is no longer frame waiting for a ack
1048 * (timeout)
1049 */
1050 if (!LIST_ISEMPTY(&SPOE_APPCTX(appctx)->waiting_queue) ||
1051 SPOE_APPCTX(appctx)->frag_ctx.ctx)
1052 return -1;
1053 appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1054 SPOE_APPCTX(appctx)->cur_fpa = 0;
1055 }
1056 return 0;
1057
1058 found:
1059 if (!spoe_acquire_buffer(&SPOE_APPCTX(appctx)->buffer,
1060 &SPOE_APPCTX(appctx)->buffer_wait)) {
1061 *ctx = NULL;
1062 return 1; /* Retry later */
1063 }
1064
1065 /* Copy encoded actions */
1066 len = (end - p);
1067 memcpy(b_head(&SPOE_APPCTX(appctx)->buffer), p, len);
1068 b_set_data(&SPOE_APPCTX(appctx)->buffer, len);
1069 p += len;
1070
1071 /* Transfer the buffer ownership to the SPOE context */
1072 (*ctx)->buffer = SPOE_APPCTX(appctx)->buffer;
1073 SPOE_APPCTX(appctx)->buffer = BUF_NULL;
1074
1075 (*ctx)->state = SPOE_CTX_ST_DONE;
1076
1077 end:
1078 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1079 " - ACK frame received"
1080 " - ctx=%p - stream-id=%u - frame-id=%u - flags=0x%08x\n",
1081 (int)now.tv_sec, (int)now.tv_usec, agent->id,
1082 __FUNCTION__, appctx, *ctx, (*ctx)->stream_id,
1083 (*ctx)->frame_id, flags);
1084 return (p - frame);
1085 }
1086
1087 /* This function is used in cfgparse.c and declared in proto/checks.h. It
1088 * prepare the request to send to agents during a healthcheck. It returns 0 on
1089 * success and -1 if an error occurred. */
1090 int
spoe_prepare_healthcheck_request(char ** req,int * len)1091 spoe_prepare_healthcheck_request(char **req, int *len)
1092 {
1093 struct appctx appctx;
1094 struct spoe_appctx spoe_appctx;
1095 char *frame, *end, buf[MAX_FRAME_SIZE+4];
1096 size_t sz;
1097 int ret;
1098
1099 memset(&appctx, 0, sizeof(appctx));
1100 memset(&spoe_appctx, 0, sizeof(spoe_appctx));
1101 memset(buf, 0, sizeof(buf));
1102
1103 appctx.ctx.spoe.ptr = &spoe_appctx;
1104 SPOE_APPCTX(&appctx)->max_frame_size = MAX_FRAME_SIZE;
1105
1106 frame = buf+4; /* Reserved the 4 first bytes for the frame size */
1107 end = frame + MAX_FRAME_SIZE;
1108
1109 ret = spoe_prepare_hahello_frame(&appctx, frame, MAX_FRAME_SIZE);
1110 if (ret <= 0)
1111 return -1;
1112 frame += ret;
1113
1114 /* Add "healthcheck" K/V item */
1115 sz = SLEN(HEALTHCHECK_KEY);
1116 if (spoe_encode_buffer(HEALTHCHECK_KEY, sz, &frame, end) == -1)
1117 return -1;
1118 *frame++ = (SPOE_DATA_T_BOOL | SPOE_DATA_FL_TRUE);
1119
1120 *len = frame - buf;
1121 sz = htonl(*len - 4);
1122 memcpy(buf, (char *)&sz, 4);
1123
1124 if ((*req = malloc(*len)) == NULL)
1125 return -1;
1126 memcpy(*req, buf, *len);
1127 return 0;
1128 }
1129
1130 /* This function is used in checks.c and declared in proto/checks.h. It decode
1131 * the response received from an agent during a healthcheck. It returns 0 on
1132 * success and -1 if an error occurred. */
1133 int
spoe_handle_healthcheck_response(char * frame,size_t size,char * err,int errlen)1134 spoe_handle_healthcheck_response(char *frame, size_t size, char *err, int errlen)
1135 {
1136 struct appctx appctx;
1137 struct spoe_appctx spoe_appctx;
1138
1139 memset(&appctx, 0, sizeof(appctx));
1140 memset(&spoe_appctx, 0, sizeof(spoe_appctx));
1141
1142 appctx.ctx.spoe.ptr = &spoe_appctx;
1143 SPOE_APPCTX(&appctx)->max_frame_size = MAX_FRAME_SIZE;
1144
1145 if (*frame == SPOE_FRM_T_AGENT_DISCON) {
1146 spoe_handle_agentdiscon_frame(&appctx, frame, size);
1147 goto error;
1148 }
1149 if (spoe_handle_agenthello_frame(&appctx, frame, size) <= 0)
1150 goto error;
1151
1152 return 0;
1153
1154 error:
1155 if (SPOE_APPCTX(&appctx)->status_code >= SPOE_FRM_ERRS)
1156 SPOE_APPCTX(&appctx)->status_code = SPOE_FRM_ERR_UNKNOWN;
1157 strncpy(err, spoe_frm_err_reasons[SPOE_APPCTX(&appctx)->status_code], errlen);
1158 return -1;
1159 }
1160
1161 /* Send a SPOE frame to an agent. It returns -1 when an error occurred, 0 when
1162 * the frame can be ignored, 1 to retry later, and the frame legnth on
1163 * success. */
1164 static int
spoe_send_frame(struct appctx * appctx,char * buf,size_t framesz)1165 spoe_send_frame(struct appctx *appctx, char *buf, size_t framesz)
1166 {
1167 struct stream_interface *si = appctx->owner;
1168 int ret;
1169 uint32_t netint;
1170
1171 /* 4 bytes are reserved at the beginning of <buf> to store the frame
1172 * length. */
1173 netint = htonl(framesz);
1174 memcpy(buf, (char *)&netint, 4);
1175 ret = ci_putblk(si_ic(si), buf, framesz+4);
1176 if (ret <= 0) {
1177 if ((ret == -3 && b_is_null(&si_ic(si)->buf)) || ret == -1) {
1178 si_rx_room_blk(si);
1179 return 1; /* retry */
1180 }
1181 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1182 return -1; /* error */
1183 }
1184 return framesz;
1185 }
1186
1187 /* Receive a SPOE frame from an agent. It return -1 when an error occurred, 0
1188 * when the frame can be ignored, 1 to retry later and the frame length on
1189 * success. */
1190 static int
spoe_recv_frame(struct appctx * appctx,char * buf,size_t framesz)1191 spoe_recv_frame(struct appctx *appctx, char *buf, size_t framesz)
1192 {
1193 struct stream_interface *si = appctx->owner;
1194 int ret;
1195 uint32_t netint;
1196
1197 ret = co_getblk(si_oc(si), (char *)&netint, 4, 0);
1198 if (ret > 0) {
1199 framesz = ntohl(netint);
1200 if (framesz > SPOE_APPCTX(appctx)->max_frame_size) {
1201 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
1202 return -1;
1203 }
1204 ret = co_getblk(si_oc(si), buf, framesz, 4);
1205 }
1206 if (ret <= 0) {
1207 if (ret == 0) {
1208 return 1; /* retry */
1209 }
1210 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1211 return -1; /* error */
1212 }
1213 return framesz;
1214 }
1215
1216 /********************************************************************
1217 * Functions that manage the SPOE applet
1218 ********************************************************************/
1219 static int
spoe_wakeup_appctx(struct appctx * appctx)1220 spoe_wakeup_appctx(struct appctx *appctx)
1221 {
1222 si_want_get(appctx->owner);
1223 si_rx_endp_more(appctx->owner);
1224 appctx_wakeup(appctx);
1225 return 1;
1226 }
1227
1228 /* Callback function that catches applet timeouts. If a timeout occurred, we set
1229 * <appctx->st1> flag and the SPOE applet is woken up. */
1230 static struct task *
spoe_process_appctx(struct task * task,void * context,unsigned short state)1231 spoe_process_appctx(struct task * task, void *context, unsigned short state)
1232 {
1233 struct appctx *appctx = context;
1234
1235 appctx->st1 = SPOE_APPCTX_ERR_NONE;
1236 if (tick_is_expired(task->expire, now_ms)) {
1237 task->expire = TICK_ETERNITY;
1238 appctx->st1 = SPOE_APPCTX_ERR_TOUT;
1239 }
1240 spoe_wakeup_appctx(appctx);
1241 return task;
1242 }
1243
1244 /* Callback function that releases a SPOE applet. This happens when the
1245 * connection with the agent is closed. */
1246 static void
spoe_release_appctx(struct appctx * appctx)1247 spoe_release_appctx(struct appctx *appctx)
1248 {
1249 struct stream_interface *si = appctx->owner;
1250 struct spoe_appctx *spoe_appctx = SPOE_APPCTX(appctx);
1251 struct spoe_agent *agent;
1252 struct spoe_context *ctx, *back;
1253
1254 if (spoe_appctx == NULL)
1255 return;
1256
1257 appctx->ctx.spoe.ptr = NULL;
1258 agent = spoe_appctx->agent;
1259
1260 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p\n",
1261 (int)now.tv_sec, (int)now.tv_usec, agent->id,
1262 __FUNCTION__, appctx);
1263
1264 /* Remove applet from the list of running applets */
1265 _HA_ATOMIC_SUB(&agent->counters.applets, 1);
1266 HA_SPIN_LOCK(SPOE_APPLET_LOCK, &agent->rt[tid].lock);
1267 if (!LIST_ISEMPTY(&spoe_appctx->list)) {
1268 LIST_DEL(&spoe_appctx->list);
1269 LIST_INIT(&spoe_appctx->list);
1270 }
1271 HA_SPIN_UNLOCK(SPOE_APPLET_LOCK, &agent->rt[tid].lock);
1272
1273 /* Shutdown the server connection, if needed */
1274 if (appctx->st0 != SPOE_APPCTX_ST_END) {
1275 if (appctx->st0 == SPOE_APPCTX_ST_IDLE) {
1276 eb32_delete(&spoe_appctx->node);
1277 _HA_ATOMIC_SUB(&agent->counters.idles, 1);
1278 }
1279
1280 appctx->st0 = SPOE_APPCTX_ST_END;
1281 if (spoe_appctx->status_code == SPOE_FRM_ERR_NONE)
1282 spoe_appctx->status_code = SPOE_FRM_ERR_IO;
1283
1284 si_shutw(si);
1285 si_shutr(si);
1286 si_ic(si)->flags |= CF_READ_NULL;
1287 }
1288
1289 /* Destroy the task attached to this applet */
1290 task_destroy(spoe_appctx->task);
1291
1292 /* Notify all waiting streams */
1293 list_for_each_entry_safe(ctx, back, &spoe_appctx->waiting_queue, list) {
1294 LIST_DEL(&ctx->list);
1295 LIST_INIT(&ctx->list);
1296 _HA_ATOMIC_SUB(&agent->counters.nb_waiting, 1);
1297 spoe_update_stat_time(&ctx->stats.tv_wait, &ctx->stats.t_waiting);
1298 ctx->spoe_appctx = NULL;
1299 ctx->state = SPOE_CTX_ST_ERROR;
1300 ctx->status_code = (spoe_appctx->status_code + 0x100);
1301 task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1302 }
1303
1304 /* If the applet was processing a fragmented frame, notify the
1305 * corresponding stream. */
1306 if (spoe_appctx->frag_ctx.ctx) {
1307 ctx = spoe_appctx->frag_ctx.ctx;
1308 ctx->spoe_appctx = NULL;
1309 ctx->state = SPOE_CTX_ST_ERROR;
1310 ctx->status_code = (spoe_appctx->status_code + 0x100);
1311 task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1312 }
1313
1314 if (!LIST_ISEMPTY(&agent->rt[tid].applets)) {
1315 list_for_each_entry_safe(ctx, back, &agent->rt[tid].waiting_queue, list) {
1316 if (ctx->spoe_appctx == spoe_appctx)
1317 ctx->spoe_appctx = NULL;
1318 }
1319 goto end;
1320 }
1321
1322 /* If this was the last running applet, notify all waiting streams */
1323 list_for_each_entry_safe(ctx, back, &agent->rt[tid].sending_queue, list) {
1324 LIST_DEL(&ctx->list);
1325 LIST_INIT(&ctx->list);
1326 _HA_ATOMIC_SUB(&agent->counters.nb_sending, 1);
1327 spoe_update_stat_time(&ctx->stats.tv_queue, &ctx->stats.t_queue);
1328 ctx->spoe_appctx = NULL;
1329 ctx->state = SPOE_CTX_ST_ERROR;
1330 ctx->status_code = (spoe_appctx->status_code + 0x100);
1331 task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1332 }
1333 list_for_each_entry_safe(ctx, back, &agent->rt[tid].waiting_queue, list) {
1334 LIST_DEL(&ctx->list);
1335 LIST_INIT(&ctx->list);
1336 _HA_ATOMIC_SUB(&agent->counters.nb_waiting, 1);
1337 spoe_update_stat_time(&ctx->stats.tv_wait, &ctx->stats.t_waiting);
1338 ctx->spoe_appctx = NULL;
1339 ctx->state = SPOE_CTX_ST_ERROR;
1340 ctx->status_code = (spoe_appctx->status_code + 0x100);
1341 task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1342 }
1343
1344 end:
1345 /* Release allocated memory */
1346 spoe_release_buffer(&spoe_appctx->buffer,
1347 &spoe_appctx->buffer_wait);
1348 pool_free(pool_head_spoe_appctx, spoe_appctx);
1349
1350 /* Update runtinme agent info */
1351 agent->rt[tid].frame_size = agent->max_frame_size;
1352 list_for_each_entry(spoe_appctx, &agent->rt[tid].applets, list)
1353 HA_ATOMIC_UPDATE_MIN(&agent->rt[tid].frame_size, spoe_appctx->max_frame_size);
1354 }
1355
1356 static int
spoe_handle_connect_appctx(struct appctx * appctx)1357 spoe_handle_connect_appctx(struct appctx *appctx)
1358 {
1359 struct stream_interface *si = appctx->owner;
1360 struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
1361 char *frame, *buf;
1362 int ret;
1363
1364 if (si_state_in(si->state, SI_SB_CER|SI_SB_DIS|SI_SB_CLO)) {
1365 /* closed */
1366 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1367 goto exit;
1368 }
1369
1370 if (!si_state_in(si->state, SI_SB_RDY|SI_SB_EST)) {
1371 /* not connected yet */
1372 si_rx_endp_more(si);
1373 task_wakeup(si_strm(si)->task, TASK_WOKEN_MSG);
1374 goto stop;
1375 }
1376
1377 if (appctx->st1 == SPOE_APPCTX_ERR_TOUT) {
1378 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1379 " - Connection timed out\n",
1380 (int)now.tv_sec, (int)now.tv_usec, agent->id,
1381 __FUNCTION__, appctx);
1382 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOUT;
1383 goto exit;
1384 }
1385
1386 if (SPOE_APPCTX(appctx)->task->expire == TICK_ETERNITY)
1387 SPOE_APPCTX(appctx)->task->expire =
1388 tick_add_ifset(now_ms, agent->timeout.hello);
1389
1390 /* 4 bytes are reserved at the beginning of <buf> to store the frame
1391 * length. */
1392 buf = trash.area; frame = buf+4;
1393 ret = spoe_prepare_hahello_frame(appctx, frame,
1394 SPOE_APPCTX(appctx)->max_frame_size);
1395 if (ret > 1)
1396 ret = spoe_send_frame(appctx, buf, ret);
1397
1398 switch (ret) {
1399 case -1: /* error */
1400 case 0: /* ignore => an error, cannot be ignored */
1401 goto exit;
1402
1403 case 1: /* retry later */
1404 goto stop;
1405
1406 default:
1407 /* HELLO frame successfully sent, now wait for the
1408 * reply. */
1409 appctx->st0 = SPOE_APPCTX_ST_CONNECTING;
1410 goto next;
1411 }
1412
1413 next:
1414 return 0;
1415 stop:
1416 return 1;
1417 exit:
1418 appctx->st0 = SPOE_APPCTX_ST_EXIT;
1419 return 0;
1420 }
1421
1422 static int
spoe_handle_connecting_appctx(struct appctx * appctx)1423 spoe_handle_connecting_appctx(struct appctx *appctx)
1424 {
1425 struct stream_interface *si = appctx->owner;
1426 struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
1427 char *frame;
1428 int ret;
1429
1430
1431 if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO) {
1432 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1433 goto exit;
1434 }
1435
1436 if (appctx->st1 == SPOE_APPCTX_ERR_TOUT) {
1437 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1438 " - Connection timed out\n",
1439 (int)now.tv_sec, (int)now.tv_usec, agent->id,
1440 __FUNCTION__, appctx);
1441 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOUT;
1442 goto exit;
1443 }
1444
1445 frame = trash.area; trash.data = 0;
1446 ret = spoe_recv_frame(appctx, frame,
1447 SPOE_APPCTX(appctx)->max_frame_size);
1448 if (ret > 1) {
1449 if (*frame == SPOE_FRM_T_AGENT_DISCON) {
1450 appctx->st0 = SPOE_APPCTX_ST_DISCONNECTING;
1451 goto next;
1452 }
1453 trash.data = ret + 4;
1454 ret = spoe_handle_agenthello_frame(appctx, frame, ret);
1455 }
1456
1457 switch (ret) {
1458 case -1: /* error */
1459 case 0: /* ignore => an error, cannot be ignored */
1460 appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1461 goto next;
1462
1463 case 1: /* retry later */
1464 goto stop;
1465
1466 default:
1467 _HA_ATOMIC_ADD(&agent->counters.idles, 1);
1468 appctx->st0 = SPOE_APPCTX_ST_IDLE;
1469 SPOE_APPCTX(appctx)->node.key = 0;
1470 eb32_insert(&agent->rt[tid].idle_applets, &SPOE_APPCTX(appctx)->node);
1471
1472 /* Update runtinme agent info */
1473 HA_ATOMIC_UPDATE_MIN(&agent->rt[tid].frame_size, SPOE_APPCTX(appctx)->max_frame_size);
1474 goto next;
1475 }
1476
1477 next:
1478 /* Do not forget to remove processed frame from the output buffer */
1479 if (trash.data)
1480 co_skip(si_oc(si), trash.data);
1481
1482 SPOE_APPCTX(appctx)->task->expire =
1483 tick_add_ifset(now_ms, agent->timeout.idle);
1484 return 0;
1485 stop:
1486 return 1;
1487 exit:
1488 appctx->st0 = SPOE_APPCTX_ST_EXIT;
1489 return 0;
1490 }
1491
1492
1493 static int
spoe_handle_sending_frame_appctx(struct appctx * appctx,int * skip)1494 spoe_handle_sending_frame_appctx(struct appctx *appctx, int *skip)
1495 {
1496 struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
1497 struct spoe_context *ctx = NULL;
1498 char *frame, *buf;
1499 int ret;
1500
1501 /* 4 bytes are reserved at the beginning of <buf> to store the frame
1502 * length. */
1503 buf = trash.area; frame = buf+4;
1504
1505 if (appctx->st0 == SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY) {
1506 ctx = SPOE_APPCTX(appctx)->frag_ctx.ctx;
1507 ret = spoe_prepare_hafrag_frame(appctx, ctx, frame,
1508 SPOE_APPCTX(appctx)->max_frame_size);
1509 }
1510 else if (LIST_ISEMPTY(&agent->rt[tid].sending_queue)) {
1511 *skip = 1;
1512 ret = 1;
1513 goto end;
1514 }
1515 else {
1516 ctx = LIST_NEXT(&agent->rt[tid].sending_queue, typeof(ctx), list);
1517 ret = spoe_prepare_hanotify_frame(appctx, ctx, frame,
1518 SPOE_APPCTX(appctx)->max_frame_size);
1519
1520 }
1521
1522 if (ret > 1)
1523 ret = spoe_send_frame(appctx, buf, ret);
1524
1525 switch (ret) {
1526 case -1: /* error */
1527 appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1528 goto end;
1529
1530 case 0: /* ignore */
1531 if (ctx == NULL)
1532 goto abort_frag_frame;
1533
1534 spoe_release_buffer(&ctx->buffer, &ctx->buffer_wait);
1535 LIST_DEL(&ctx->list);
1536 LIST_INIT(&ctx->list);
1537 _HA_ATOMIC_SUB(&agent->counters.nb_sending, 1);
1538 spoe_update_stat_time(&ctx->stats.tv_queue, &ctx->stats.t_queue);
1539 ctx->spoe_appctx = NULL;
1540 ctx->state = SPOE_CTX_ST_ERROR;
1541 ctx->status_code = (SPOE_APPCTX(appctx)->status_code + 0x100);
1542 task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1543 *skip = 1;
1544 break;
1545
1546 case 1: /* retry */
1547 *skip = 1;
1548 break;
1549
1550 default:
1551 if (ctx == NULL)
1552 goto abort_frag_frame;
1553
1554 spoe_release_buffer(&ctx->buffer, &ctx->buffer_wait);
1555 LIST_DEL(&ctx->list);
1556 LIST_INIT(&ctx->list);
1557 _HA_ATOMIC_SUB(&agent->counters.nb_sending, 1);
1558 spoe_update_stat_time(&ctx->stats.tv_queue, &ctx->stats.t_queue);
1559 ctx->spoe_appctx = SPOE_APPCTX(appctx);
1560 if (!(ctx->flags & SPOE_CTX_FL_FRAGMENTED) ||
1561 (ctx->frag_ctx.flags & SPOE_FRM_FL_FIN))
1562 goto no_frag_frame_sent;
1563 else
1564 goto frag_frame_sent;
1565 }
1566 goto end;
1567
1568 frag_frame_sent:
1569 appctx->st0 = SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY;
1570 *skip = 1;
1571 SPOE_APPCTX(appctx)->frag_ctx.ctx = ctx;
1572 SPOE_APPCTX(appctx)->frag_ctx.cursid = ctx->stream_id;
1573 SPOE_APPCTX(appctx)->frag_ctx.curfid = ctx->frame_id;
1574 ctx->state = SPOE_CTX_ST_ENCODING_MSGS;
1575 task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1576 goto end;
1577
1578 no_frag_frame_sent:
1579 if (SPOE_APPCTX(appctx)->flags & SPOE_APPCTX_FL_ASYNC) {
1580 appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1581 LIST_ADDQ(&agent->rt[tid].waiting_queue, &ctx->list);
1582 }
1583 else if (SPOE_APPCTX(appctx)->flags & SPOE_APPCTX_FL_PIPELINING) {
1584 appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1585 LIST_ADDQ(&SPOE_APPCTX(appctx)->waiting_queue, &ctx->list);
1586 }
1587 else {
1588 appctx->st0 = SPOE_APPCTX_ST_WAITING_SYNC_ACK;
1589 *skip = 1;
1590 LIST_ADDQ(&SPOE_APPCTX(appctx)->waiting_queue, &ctx->list);
1591 }
1592 _HA_ATOMIC_ADD(&agent->counters.nb_waiting, 1);
1593 ctx->stats.tv_wait = now;
1594 SPOE_APPCTX(appctx)->frag_ctx.ctx = NULL;
1595 SPOE_APPCTX(appctx)->frag_ctx.cursid = 0;
1596 SPOE_APPCTX(appctx)->frag_ctx.curfid = 0;
1597 SPOE_APPCTX(appctx)->cur_fpa++;
1598
1599 ctx->state = SPOE_CTX_ST_WAITING_ACK;
1600 goto end;
1601
1602 abort_frag_frame:
1603 appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1604 SPOE_APPCTX(appctx)->frag_ctx.ctx = NULL;
1605 SPOE_APPCTX(appctx)->frag_ctx.cursid = 0;
1606 SPOE_APPCTX(appctx)->frag_ctx.curfid = 0;
1607 goto end;
1608
1609 end:
1610 return ret;
1611 }
1612
1613 static int
spoe_handle_receiving_frame_appctx(struct appctx * appctx,int * skip)1614 spoe_handle_receiving_frame_appctx(struct appctx *appctx, int *skip)
1615 {
1616 struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
1617 struct spoe_context *ctx = NULL;
1618 char *frame;
1619 int ret;
1620
1621 frame = trash.area; trash.data = 0;
1622 ret = spoe_recv_frame(appctx, frame,
1623 SPOE_APPCTX(appctx)->max_frame_size);
1624 if (ret > 1) {
1625 if (*frame == SPOE_FRM_T_AGENT_DISCON) {
1626 appctx->st0 = SPOE_APPCTX_ST_DISCONNECTING;
1627 ret = -1;
1628 goto end;
1629 }
1630 trash.data = ret + 4;
1631 ret = spoe_handle_agentack_frame(appctx, &ctx, frame, ret);
1632 }
1633 switch (ret) {
1634 case -1: /* error */
1635 appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1636 break;
1637
1638 case 0: /* ignore */
1639 break;
1640
1641 case 1: /* retry */
1642 *skip = 1;
1643 break;
1644
1645 default:
1646 LIST_DEL(&ctx->list);
1647 LIST_INIT(&ctx->list);
1648 _HA_ATOMIC_SUB(&agent->counters.nb_waiting, 1);
1649 spoe_update_stat_time(&ctx->stats.tv_wait, &ctx->stats.t_waiting);
1650 ctx->stats.tv_response = now;
1651 if (ctx->spoe_appctx) {
1652 ctx->spoe_appctx->cur_fpa--;
1653 ctx->spoe_appctx = NULL;
1654 }
1655 if (appctx->st0 == SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY &&
1656 ctx == SPOE_APPCTX(appctx)->frag_ctx.ctx) {
1657 appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1658 SPOE_APPCTX(appctx)->frag_ctx.ctx = NULL;
1659 SPOE_APPCTX(appctx)->frag_ctx.cursid = 0;
1660 SPOE_APPCTX(appctx)->frag_ctx.curfid = 0;
1661 }
1662 else if (appctx->st0 == SPOE_APPCTX_ST_WAITING_SYNC_ACK)
1663 appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1664 task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1665 break;
1666 }
1667
1668 /* Do not forget to remove processed frame from the output buffer */
1669 if (trash.data)
1670 co_skip(si_oc(appctx->owner), trash.data);
1671 end:
1672 return ret;
1673 }
1674
1675 static int
spoe_handle_processing_appctx(struct appctx * appctx)1676 spoe_handle_processing_appctx(struct appctx *appctx)
1677 {
1678 struct stream_interface *si = appctx->owner;
1679 struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
1680 int ret, skip_sending = 0, skip_receiving = 0, active_s = 0, active_r = 0;
1681
1682 if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO) {
1683 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1684 goto exit;
1685 }
1686
1687 if (appctx->st1 == SPOE_APPCTX_ERR_TOUT) {
1688 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOUT;
1689 appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1690 appctx->st1 = SPOE_APPCTX_ERR_NONE;
1691 goto next;
1692 }
1693
1694 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1695 " - process: fpa=%u/%u - appctx-state=%s - weight=%u - flags=0x%08x\n",
1696 (int)now.tv_sec, (int)now.tv_usec, agent->id,
1697 __FUNCTION__, appctx, SPOE_APPCTX(appctx)->cur_fpa,
1698 agent->max_fpa, spoe_appctx_state_str[appctx->st0],
1699 SPOE_APPCTX(appctx)->node.key, SPOE_APPCTX(appctx)->flags);
1700
1701 if (appctx->st0 == SPOE_APPCTX_ST_WAITING_SYNC_ACK)
1702 skip_sending = 1;
1703
1704 /* receiving_frame loop */
1705 while (!skip_receiving) {
1706 ret = spoe_handle_receiving_frame_appctx(appctx, &skip_receiving);
1707 switch (ret) {
1708 case -1: /* error */
1709 goto next;
1710
1711 case 0: /* ignore */
1712 active_r = 1;
1713 break;
1714
1715 case 1: /* retry */
1716 break;
1717
1718 default:
1719 active_r = 1;
1720 break;
1721 }
1722 }
1723
1724 /* send_frame loop */
1725 while (!skip_sending && SPOE_APPCTX(appctx)->cur_fpa < agent->max_fpa) {
1726 ret = spoe_handle_sending_frame_appctx(appctx, &skip_sending);
1727 switch (ret) {
1728 case -1: /* error */
1729 goto next;
1730
1731 case 0: /* ignore */
1732 if (SPOE_APPCTX(appctx)->node.key)
1733 SPOE_APPCTX(appctx)->node.key--;
1734 active_s++;
1735 break;
1736
1737 case 1: /* retry */
1738 break;
1739
1740 default:
1741 if (SPOE_APPCTX(appctx)->node.key)
1742 SPOE_APPCTX(appctx)->node.key--;
1743 active_s++;
1744 break;
1745 }
1746 }
1747
1748 if (active_s || active_r) {
1749 update_freq_ctr(&agent->rt[tid].processing_per_sec, active_s);
1750 SPOE_APPCTX(appctx)->task->expire = tick_add_ifset(now_ms, agent->timeout.idle);
1751 }
1752
1753 if (appctx->st0 == SPOE_APPCTX_ST_PROCESSING && SPOE_APPCTX(appctx)->cur_fpa < agent->max_fpa) {
1754 struct server *srv = objt_server(si_strm(si)->target);
1755
1756 /* With several threads, close the applet if there are pending
1757 * connections or if the server is full. Otherwise, add the
1758 * applet in the idle list.
1759 */
1760 if (global.nbthread > 1 &&
1761 (agent->b.be->nbpend ||
1762 (srv && (srv->nbpend || (srv->maxconn && srv->served >=srv_dynamic_maxconn(srv)))))) {
1763 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NONE;
1764 appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1765 appctx->st1 = SPOE_APPCTX_ERR_NONE;
1766 goto next;
1767 }
1768 _HA_ATOMIC_ADD(&agent->counters.idles, 1);
1769 appctx->st0 = SPOE_APPCTX_ST_IDLE;
1770 eb32_insert(&agent->rt[tid].idle_applets, &SPOE_APPCTX(appctx)->node);
1771 }
1772 return 1;
1773
1774 next:
1775 SPOE_APPCTX(appctx)->task->expire = tick_add_ifset(now_ms, agent->timeout.idle);
1776 return 0;
1777
1778 exit:
1779 appctx->st0 = SPOE_APPCTX_ST_EXIT;
1780 return 0;
1781 }
1782
1783 static int
spoe_handle_disconnect_appctx(struct appctx * appctx)1784 spoe_handle_disconnect_appctx(struct appctx *appctx)
1785 {
1786 struct stream_interface *si = appctx->owner;
1787 struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
1788 char *frame, *buf;
1789 int ret;
1790
1791 if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO)
1792 goto exit;
1793
1794 if (appctx->st1 == SPOE_APPCTX_ERR_TOUT)
1795 goto exit;
1796
1797 /* 4 bytes are reserved at the beginning of <buf> to store the frame
1798 * length. */
1799 buf = trash.area; frame = buf+4;
1800 ret = spoe_prepare_hadiscon_frame(appctx, frame,
1801 SPOE_APPCTX(appctx)->max_frame_size);
1802 if (ret > 1)
1803 ret = spoe_send_frame(appctx, buf, ret);
1804
1805 switch (ret) {
1806 case -1: /* error */
1807 case 0: /* ignore => an error, cannot be ignored */
1808 goto exit;
1809
1810 case 1: /* retry */
1811 goto stop;
1812
1813 default:
1814 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1815 " - disconnected by HAProxy (%d): %s\n",
1816 (int)now.tv_sec, (int)now.tv_usec, agent->id,
1817 __FUNCTION__, appctx,
1818 SPOE_APPCTX(appctx)->status_code,
1819 spoe_frm_err_reasons[SPOE_APPCTX(appctx)->status_code]);
1820
1821 appctx->st0 = SPOE_APPCTX_ST_DISCONNECTING;
1822 goto next;
1823 }
1824
1825 next:
1826 SPOE_APPCTX(appctx)->task->expire =
1827 tick_add_ifset(now_ms, agent->timeout.idle);
1828 return 0;
1829 stop:
1830 return 1;
1831 exit:
1832 appctx->st0 = SPOE_APPCTX_ST_EXIT;
1833 return 0;
1834 }
1835
1836 static int
spoe_handle_disconnecting_appctx(struct appctx * appctx)1837 spoe_handle_disconnecting_appctx(struct appctx *appctx)
1838 {
1839 struct stream_interface *si = appctx->owner;
1840 char *frame;
1841 int ret;
1842
1843 if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO) {
1844 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1845 goto exit;
1846 }
1847
1848 if (appctx->st1 == SPOE_APPCTX_ERR_TOUT) {
1849 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOUT;
1850 goto exit;
1851 }
1852
1853 frame = trash.area; trash.data = 0;
1854 ret = spoe_recv_frame(appctx, frame,
1855 SPOE_APPCTX(appctx)->max_frame_size);
1856 if (ret > 1) {
1857 trash.data = ret + 4;
1858 ret = spoe_handle_agentdiscon_frame(appctx, frame, ret);
1859 }
1860
1861 switch (ret) {
1862 case -1: /* error */
1863 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1864 " - error on frame (%s)\n",
1865 (int)now.tv_sec, (int)now.tv_usec,
1866 ((struct spoe_agent *)SPOE_APPCTX(appctx)->agent)->id,
1867 __FUNCTION__, appctx,
1868 spoe_frm_err_reasons[SPOE_APPCTX(appctx)->status_code]);
1869 goto exit;
1870
1871 case 0: /* ignore */
1872 goto next;
1873
1874 case 1: /* retry */
1875 goto stop;
1876
1877 default:
1878 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1879 " - disconnected by peer (%d): %.*s\n",
1880 (int)now.tv_sec, (int)now.tv_usec,
1881 ((struct spoe_agent *)SPOE_APPCTX(appctx)->agent)->id,
1882 __FUNCTION__, appctx, SPOE_APPCTX(appctx)->status_code,
1883 SPOE_APPCTX(appctx)->rlen, SPOE_APPCTX(appctx)->reason);
1884 goto exit;
1885 }
1886
1887 next:
1888 /* Do not forget to remove processed frame from the output buffer */
1889 if (trash.data)
1890 co_skip(si_oc(appctx->owner), trash.data);
1891
1892 return 0;
1893 stop:
1894 return 1;
1895 exit:
1896 appctx->st0 = SPOE_APPCTX_ST_EXIT;
1897 return 0;
1898 }
1899
1900 /* I/O Handler processing messages exchanged with the agent */
1901 static void
spoe_handle_appctx(struct appctx * appctx)1902 spoe_handle_appctx(struct appctx *appctx)
1903 {
1904 struct stream_interface *si = appctx->owner;
1905 struct spoe_agent *agent;
1906
1907 if (SPOE_APPCTX(appctx) == NULL)
1908 return;
1909
1910 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NONE;
1911 agent = SPOE_APPCTX(appctx)->agent;
1912
1913 switchstate:
1914 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1915 " - appctx-state=%s\n",
1916 (int)now.tv_sec, (int)now.tv_usec, agent->id,
1917 __FUNCTION__, appctx, spoe_appctx_state_str[appctx->st0]);
1918
1919 switch (appctx->st0) {
1920 case SPOE_APPCTX_ST_CONNECT:
1921 if (spoe_handle_connect_appctx(appctx))
1922 goto out;
1923 goto switchstate;
1924
1925 case SPOE_APPCTX_ST_CONNECTING:
1926 if (spoe_handle_connecting_appctx(appctx))
1927 goto out;
1928 goto switchstate;
1929
1930 case SPOE_APPCTX_ST_IDLE:
1931 _HA_ATOMIC_SUB(&agent->counters.idles, 1);
1932 eb32_delete(&SPOE_APPCTX(appctx)->node);
1933 if (stopping &&
1934 LIST_ISEMPTY(&agent->rt[tid].sending_queue) &&
1935 LIST_ISEMPTY(&SPOE_APPCTX(appctx)->waiting_queue)) {
1936 SPOE_APPCTX(appctx)->task->expire =
1937 tick_add_ifset(now_ms, agent->timeout.idle);
1938 appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1939 goto switchstate;
1940 }
1941 appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1942 /* fall through */
1943
1944 case SPOE_APPCTX_ST_PROCESSING:
1945 case SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY:
1946 case SPOE_APPCTX_ST_WAITING_SYNC_ACK:
1947 if (spoe_handle_processing_appctx(appctx))
1948 goto out;
1949 goto switchstate;
1950
1951 case SPOE_APPCTX_ST_DISCONNECT:
1952 if (spoe_handle_disconnect_appctx(appctx))
1953 goto out;
1954 goto switchstate;
1955
1956 case SPOE_APPCTX_ST_DISCONNECTING:
1957 if (spoe_handle_disconnecting_appctx(appctx))
1958 goto out;
1959 goto switchstate;
1960
1961 case SPOE_APPCTX_ST_EXIT:
1962 appctx->st0 = SPOE_APPCTX_ST_END;
1963 SPOE_APPCTX(appctx)->task->expire = TICK_ETERNITY;
1964
1965 si_shutw(si);
1966 si_shutr(si);
1967 si_ic(si)->flags |= CF_READ_NULL;
1968 /* fall through */
1969
1970 case SPOE_APPCTX_ST_END:
1971 return;
1972 }
1973 out:
1974 if (stopping)
1975 spoe_wakeup_appctx(appctx);
1976
1977 if (SPOE_APPCTX(appctx)->task->expire != TICK_ETERNITY)
1978 task_queue(SPOE_APPCTX(appctx)->task);
1979 }
1980
1981 struct applet spoe_applet = {
1982 .obj_type = OBJ_TYPE_APPLET,
1983 .name = "<SPOE>", /* used for logging */
1984 .fct = spoe_handle_appctx,
1985 .release = spoe_release_appctx,
1986 };
1987
1988 /* Create a SPOE applet. On success, the created applet is returned, else
1989 * NULL. */
1990 static struct appctx *
spoe_create_appctx(struct spoe_config * conf)1991 spoe_create_appctx(struct spoe_config *conf)
1992 {
1993 struct appctx *appctx;
1994 struct session *sess;
1995 struct stream *strm;
1996
1997 if ((appctx = appctx_new(&spoe_applet, tid_bit)) == NULL)
1998 goto out_error;
1999
2000 appctx->ctx.spoe.ptr = pool_alloc_dirty(pool_head_spoe_appctx);
2001 if (SPOE_APPCTX(appctx) == NULL)
2002 goto out_free_appctx;
2003 memset(appctx->ctx.spoe.ptr, 0, pool_head_spoe_appctx->size);
2004
2005 appctx->st0 = SPOE_APPCTX_ST_CONNECT;
2006 if ((SPOE_APPCTX(appctx)->task = task_new(tid_bit)) == NULL)
2007 goto out_free_spoe_appctx;
2008
2009 SPOE_APPCTX(appctx)->owner = appctx;
2010 SPOE_APPCTX(appctx)->task->process = spoe_process_appctx;
2011 SPOE_APPCTX(appctx)->task->context = appctx;
2012 SPOE_APPCTX(appctx)->agent = conf->agent;
2013 SPOE_APPCTX(appctx)->version = 0;
2014 SPOE_APPCTX(appctx)->max_frame_size = conf->agent->max_frame_size;
2015 SPOE_APPCTX(appctx)->flags = 0;
2016 SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NONE;
2017 SPOE_APPCTX(appctx)->buffer = BUF_NULL;
2018 SPOE_APPCTX(appctx)->cur_fpa = 0;
2019
2020 LIST_INIT(&SPOE_APPCTX(appctx)->buffer_wait.list);
2021 SPOE_APPCTX(appctx)->buffer_wait.target = appctx;
2022 SPOE_APPCTX(appctx)->buffer_wait.wakeup_cb = (int (*)(void *))spoe_wakeup_appctx;
2023
2024 LIST_INIT(&SPOE_APPCTX(appctx)->list);
2025 LIST_INIT(&SPOE_APPCTX(appctx)->waiting_queue);
2026
2027 sess = session_new(&conf->agent_fe, NULL, &appctx->obj_type);
2028 if (!sess)
2029 goto out_free_spoe;
2030
2031 if ((strm = stream_new(sess, &appctx->obj_type)) == NULL)
2032 goto out_free_sess;
2033
2034 stream_set_backend(strm, conf->agent->b.be);
2035
2036 /* applet is waiting for data */
2037 si_cant_get(&strm->si[0]);
2038 appctx_wakeup(appctx);
2039
2040 strm->do_log = NULL;
2041 strm->res.flags |= CF_READ_DONTWAIT;
2042
2043 HA_SPIN_LOCK(SPOE_APPLET_LOCK, &conf->agent->rt[tid].lock);
2044 LIST_ADDQ(&conf->agent->rt[tid].applets, &SPOE_APPCTX(appctx)->list);
2045 HA_SPIN_UNLOCK(SPOE_APPLET_LOCK, &conf->agent->rt[tid].lock);
2046 _HA_ATOMIC_ADD(&conf->agent->counters.applets, 1);
2047
2048 task_wakeup(SPOE_APPCTX(appctx)->task, TASK_WOKEN_INIT);
2049 task_wakeup(strm->task, TASK_WOKEN_INIT);
2050 return appctx;
2051
2052 /* Error unrolling */
2053 out_free_sess:
2054 session_free(sess);
2055 out_free_spoe:
2056 task_destroy(SPOE_APPCTX(appctx)->task);
2057 out_free_spoe_appctx:
2058 pool_free(pool_head_spoe_appctx, SPOE_APPCTX(appctx));
2059 out_free_appctx:
2060 appctx_free(appctx);
2061 out_error:
2062 return NULL;
2063 }
2064
2065 static int
spoe_queue_context(struct spoe_context * ctx)2066 spoe_queue_context(struct spoe_context *ctx)
2067 {
2068 struct spoe_config *conf = FLT_CONF(ctx->filter);
2069 struct spoe_agent *agent = conf->agent;
2070 struct appctx *appctx;
2071 struct spoe_appctx *spoe_appctx;
2072
2073 /* Check if we need to create a new SPOE applet or not. */
2074 if (!eb_is_empty(&agent->rt[tid].idle_applets) &&
2075 (agent->rt[tid].processing == 1 || agent->rt[tid].processing < read_freq_ctr(&agent->rt[tid].processing_per_sec)))
2076 goto end;
2077
2078 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2079 " - try to create new SPOE appctx\n",
2080 (int)now.tv_sec, (int)now.tv_usec, agent->id, __FUNCTION__,
2081 ctx->strm);
2082
2083 /* Do not try to create a new applet if there is no server up for the
2084 * agent's backend. */
2085 if (!agent->b.be->srv_act && !agent->b.be->srv_bck) {
2086 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2087 " - cannot create SPOE appctx: no server up\n",
2088 (int)now.tv_sec, (int)now.tv_usec, agent->id,
2089 __FUNCTION__, ctx->strm);
2090 goto end;
2091 }
2092
2093 /* Do not try to create a new applet if we have reached the maximum of
2094 * connection per seconds */
2095 if (agent->cps_max > 0) {
2096 if (!freq_ctr_remain(&agent->rt[tid].conn_per_sec, agent->cps_max, 0)) {
2097 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2098 " - cannot create SPOE appctx: max CPS reached\n",
2099 (int)now.tv_sec, (int)now.tv_usec, agent->id,
2100 __FUNCTION__, ctx->strm);
2101 goto end;
2102 }
2103 }
2104
2105 appctx = spoe_create_appctx(conf);
2106 if (appctx == NULL) {
2107 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2108 " - failed to create SPOE appctx\n",
2109 (int)now.tv_sec, (int)now.tv_usec, agent->id,
2110 __FUNCTION__, ctx->strm);
2111 send_log(&conf->agent_fe, LOG_EMERG,
2112 "SPOE: [%s] failed to create SPOE applet\n",
2113 agent->id);
2114
2115 goto end;
2116 }
2117
2118 /* Increase the per-process number of cumulated connections */
2119 if (agent->cps_max > 0)
2120 update_freq_ctr(&agent->rt[tid].conn_per_sec, 1);
2121
2122 end:
2123 /* The only reason to return an error is when there is no applet */
2124 if (LIST_ISEMPTY(&agent->rt[tid].applets)) {
2125 ctx->status_code = SPOE_CTX_ERR_RES;
2126 return -1;
2127 }
2128
2129 /* Add the SPOE context in the sending queue if the stream has no applet
2130 * already assigned and wakeup all idle applets. Otherwise, don't queue
2131 * it. */
2132 _HA_ATOMIC_ADD(&agent->counters.nb_sending, 1);
2133 spoe_update_stat_time(&ctx->stats.tv_request, &ctx->stats.t_request);
2134 ctx->stats.tv_queue = now;
2135 if (ctx->spoe_appctx)
2136 return 1;
2137 LIST_ADDQ(&agent->rt[tid].sending_queue, &ctx->list);
2138
2139 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2140 " - Add stream in sending queue"
2141 " - applets=%u - idles=%u - processing=%u\n",
2142 (int)now.tv_sec, (int)now.tv_usec, agent->id, __FUNCTION__,
2143 ctx->strm, agent->counters.applets, agent->counters.idles,
2144 agent->rt[tid].processing);
2145
2146 /* Finally try to wakeup an IDLE applet. */
2147 if (!eb_is_empty(&agent->rt[tid].idle_applets)) {
2148 struct eb32_node *node;
2149
2150 node = eb32_first(&agent->rt[tid].idle_applets);
2151 spoe_appctx = eb32_entry(node, struct spoe_appctx, node);
2152 if (node && spoe_appctx) {
2153 eb32_delete(&spoe_appctx->node);
2154 spoe_appctx->node.key++;
2155 eb32_insert(&agent->rt[tid].idle_applets, &spoe_appctx->node);
2156 spoe_wakeup_appctx(spoe_appctx->owner);
2157 }
2158 }
2159 return 1;
2160 }
2161
2162 /***************************************************************************
2163 * Functions that encode SPOE messages
2164 **************************************************************************/
2165 /* Encode a SPOE message. Info in <ctx->frag_ctx>, if any, are used to handle
2166 * fragmented_content. If the next message can be processed, it returns 0. If
2167 * the message is too big, it returns -1.*/
2168 static int
spoe_encode_message(struct stream * s,struct spoe_context * ctx,struct spoe_message * msg,int dir,char ** buf,char * end)2169 spoe_encode_message(struct stream *s, struct spoe_context *ctx,
2170 struct spoe_message *msg, int dir,
2171 char **buf, char *end)
2172 {
2173 struct sample *smp;
2174 struct spoe_arg *arg;
2175 int ret;
2176
2177 if (msg->cond) {
2178 ret = acl_exec_cond(msg->cond, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2179 ret = acl_pass(ret);
2180 if (msg->cond->pol == ACL_COND_UNLESS)
2181 ret = !ret;
2182
2183 /* the rule does not match */
2184 if (!ret)
2185 goto next;
2186 }
2187
2188 /* Resume encoding of a SPOE argument */
2189 if (ctx->frag_ctx.curarg != NULL) {
2190 arg = ctx->frag_ctx.curarg;
2191 goto encode_argument;
2192 }
2193
2194 if (ctx->frag_ctx.curoff != UINT_MAX)
2195 goto encode_msg_payload;
2196
2197 /* Check if there is enough space for the message name and the
2198 * number of arguments. It implies <msg->id_len> is encoded on 2
2199 * bytes, at most (< 2288). */
2200 if (*buf + 2 + msg->id_len + 1 > end)
2201 goto too_big;
2202
2203 /* Encode the message name */
2204 if (spoe_encode_buffer(msg->id, msg->id_len, buf, end) == -1)
2205 goto too_big;
2206
2207 /* Set the number of arguments for this message */
2208 **buf = msg->nargs;
2209 (*buf)++;
2210
2211 ctx->frag_ctx.curoff = 0;
2212 encode_msg_payload:
2213
2214 /* Loop on arguments */
2215 list_for_each_entry(arg, &msg->args, list) {
2216 ctx->frag_ctx.curarg = arg;
2217 ctx->frag_ctx.curoff = UINT_MAX;
2218 ctx->frag_ctx.curlen = 0;
2219
2220 encode_argument:
2221 if (ctx->frag_ctx.curoff != UINT_MAX)
2222 goto encode_arg_value;
2223
2224 /* Encode the argument name as a string. It can by NULL */
2225 if (spoe_encode_buffer(arg->name, arg->name_len, buf, end) == -1)
2226 goto too_big;
2227
2228 ctx->frag_ctx.curoff = 0;
2229 encode_arg_value:
2230
2231 /* Fetch the argument value */
2232 smp = sample_process(s->be, s->sess, s, dir|SMP_OPT_FINAL, arg->expr, NULL);
2233 if (smp) {
2234 smp->ctx.a[0] = &ctx->frag_ctx.curlen;
2235 smp->ctx.a[1] = &ctx->frag_ctx.curoff;
2236 }
2237 ret = spoe_encode_data(smp, buf, end);
2238 if (ret == -1 || ctx->frag_ctx.curoff)
2239 goto too_big;
2240 }
2241
2242 next:
2243 return 0;
2244
2245 too_big:
2246 return -1;
2247 }
2248
2249 /* Encode list of SPOE messages. Info in <ctx->frag_ctx>, if any, are used to
2250 * handle fragmented content. On success it returns 1. If an error occurred, -1
2251 * is returned. If nothing has been encoded, it returns 0 (this is only possible
2252 * for unfragmented payload). */
2253 static int
spoe_encode_messages(struct stream * s,struct spoe_context * ctx,struct list * messages,int dir,int type)2254 spoe_encode_messages(struct stream *s, struct spoe_context *ctx,
2255 struct list *messages, int dir, int type)
2256 {
2257 struct spoe_config *conf = FLT_CONF(ctx->filter);
2258 struct spoe_agent *agent = conf->agent;
2259 struct spoe_message *msg;
2260 char *p, *end;
2261
2262 p = b_head(&ctx->buffer);
2263 end = p + agent->rt[tid].frame_size - FRAME_HDR_SIZE;
2264
2265 if (type == SPOE_MSGS_BY_EVENT) { /* Loop on messages by event */
2266 /* Resume encoding of a SPOE message */
2267 if (ctx->frag_ctx.curmsg != NULL) {
2268 msg = ctx->frag_ctx.curmsg;
2269 goto encode_evt_message;
2270 }
2271
2272 list_for_each_entry(msg, messages, by_evt) {
2273 ctx->frag_ctx.curmsg = msg;
2274 ctx->frag_ctx.curarg = NULL;
2275 ctx->frag_ctx.curoff = UINT_MAX;
2276
2277 encode_evt_message:
2278 if (spoe_encode_message(s, ctx, msg, dir, &p, end) == -1)
2279 goto too_big;
2280 }
2281 }
2282 else if (type == SPOE_MSGS_BY_GROUP) { /* Loop on messages by group */
2283 /* Resume encoding of a SPOE message */
2284 if (ctx->frag_ctx.curmsg != NULL) {
2285 msg = ctx->frag_ctx.curmsg;
2286 goto encode_grp_message;
2287 }
2288
2289 list_for_each_entry(msg, messages, by_grp) {
2290 ctx->frag_ctx.curmsg = msg;
2291 ctx->frag_ctx.curarg = NULL;
2292 ctx->frag_ctx.curoff = UINT_MAX;
2293
2294 encode_grp_message:
2295 if (spoe_encode_message(s, ctx, msg, dir, &p, end) == -1)
2296 goto too_big;
2297 }
2298 }
2299 else
2300 goto skip;
2301
2302
2303 /* nothing has been encoded for an unfragmented payload */
2304 if (!(ctx->flags & SPOE_CTX_FL_FRAGMENTED) && p == b_head(&ctx->buffer))
2305 goto skip;
2306
2307 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2308 " - encode %s messages - spoe_appctx=%p"
2309 "- max_size=%u - encoded=%ld\n",
2310 (int)now.tv_sec, (int)now.tv_usec,
2311 agent->id, __FUNCTION__, s,
2312 ((ctx->flags & SPOE_CTX_FL_FRAGMENTED) ? "last fragment of" : "unfragmented"),
2313 ctx->spoe_appctx, (agent->rt[tid].frame_size - FRAME_HDR_SIZE),
2314 p - b_head(&ctx->buffer));
2315
2316 b_set_data(&ctx->buffer, p - b_head(&ctx->buffer));
2317 ctx->frag_ctx.curmsg = NULL;
2318 ctx->frag_ctx.curarg = NULL;
2319 ctx->frag_ctx.curoff = 0;
2320 ctx->frag_ctx.flags = SPOE_FRM_FL_FIN;
2321
2322 return 1;
2323
2324 too_big:
2325 /* Return an error if fragmentation is unsupported or if nothing has
2326 * been encoded because its too big and not splittable. */
2327 if (!(agent->flags & SPOE_FL_SND_FRAGMENTATION) || p == b_head(&ctx->buffer)) {
2328 ctx->status_code = SPOE_CTX_ERR_TOO_BIG;
2329 return -1;
2330 }
2331
2332 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2333 " - encode fragmented messages - spoe_appctx=%p"
2334 " - curmsg=%p - curarg=%p - curoff=%u"
2335 " - max_size=%u - encoded=%ld\n",
2336 (int)now.tv_sec, (int)now.tv_usec,
2337 agent->id, __FUNCTION__, s, ctx->spoe_appctx,
2338 ctx->frag_ctx.curmsg, ctx->frag_ctx.curarg, ctx->frag_ctx.curoff,
2339 (agent->rt[tid].frame_size - FRAME_HDR_SIZE), p - b_head(&ctx->buffer));
2340
2341 b_set_data(&ctx->buffer, p - b_head(&ctx->buffer));
2342 ctx->flags |= SPOE_CTX_FL_FRAGMENTED;
2343 ctx->frag_ctx.flags &= ~SPOE_FRM_FL_FIN;
2344 return 1;
2345
2346 skip:
2347 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2348 " - skip the frame because nothing has been encoded\n",
2349 (int)now.tv_sec, (int)now.tv_usec,
2350 agent->id, __FUNCTION__, s);
2351 return 0;
2352 }
2353
2354
2355 /***************************************************************************
2356 * Functions that handle SPOE actions
2357 **************************************************************************/
2358 /* Helper function to set a variable */
2359 static void
spoe_set_var(struct spoe_context * ctx,char * scope,char * name,int len,struct sample * smp)2360 spoe_set_var(struct spoe_context *ctx, char *scope, char *name, int len,
2361 struct sample *smp)
2362 {
2363 struct spoe_config *conf = FLT_CONF(ctx->filter);
2364 struct spoe_agent *agent = conf->agent;
2365 char varname[64];
2366
2367 memset(varname, 0, sizeof(varname));
2368 len = snprintf(varname, sizeof(varname), "%s.%s.%.*s",
2369 scope, agent->var_pfx, len, name);
2370 if (agent->flags & SPOE_FL_FORCE_SET_VAR)
2371 vars_set_by_name(varname, len, smp);
2372 else
2373 vars_set_by_name_ifexist(varname, len, smp);
2374 }
2375
2376 /* Helper function to unset a variable */
2377 static void
spoe_unset_var(struct spoe_context * ctx,char * scope,char * name,int len,struct sample * smp)2378 spoe_unset_var(struct spoe_context *ctx, char *scope, char *name, int len,
2379 struct sample *smp)
2380 {
2381 struct spoe_config *conf = FLT_CONF(ctx->filter);
2382 struct spoe_agent *agent = conf->agent;
2383 char varname[64];
2384
2385 memset(varname, 0, sizeof(varname));
2386 len = snprintf(varname, sizeof(varname), "%s.%s.%.*s",
2387 scope, agent->var_pfx, len, name);
2388 vars_unset_by_name_ifexist(varname, len, smp);
2389 }
2390
2391
2392 static inline int
spoe_decode_action_set_var(struct stream * s,struct spoe_context * ctx,char ** buf,char * end,int dir)2393 spoe_decode_action_set_var(struct stream *s, struct spoe_context *ctx,
2394 char **buf, char *end, int dir)
2395 {
2396 char *str, *scope, *p = *buf;
2397 struct sample smp;
2398 uint64_t sz;
2399 int ret;
2400
2401 if (p + 2 >= end)
2402 goto skip;
2403
2404 /* SET-VAR requires 3 arguments */
2405 if (*p++ != 3)
2406 goto skip;
2407
2408 switch (*p++) {
2409 case SPOE_SCOPE_PROC: scope = "proc"; break;
2410 case SPOE_SCOPE_SESS: scope = "sess"; break;
2411 case SPOE_SCOPE_TXN : scope = "txn"; break;
2412 case SPOE_SCOPE_REQ : scope = "req"; break;
2413 case SPOE_SCOPE_RES : scope = "res"; break;
2414 default: goto skip;
2415 }
2416
2417 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
2418 goto skip;
2419 memset(&smp, 0, sizeof(smp));
2420 smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2421
2422 if (spoe_decode_data(&p, end, &smp) == -1)
2423 goto skip;
2424
2425 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2426 " - set-var '%s.%s.%.*s'\n",
2427 (int)now.tv_sec, (int)now.tv_usec,
2428 ((struct spoe_config *)FLT_CONF(ctx->filter))->agent->id,
2429 __FUNCTION__, s, scope,
2430 ((struct spoe_config *)FLT_CONF(ctx->filter))->agent->var_pfx,
2431 (int)sz, str);
2432
2433 if (smp.data.type == SMP_T_ANY)
2434 spoe_unset_var(ctx, scope, str, sz, &smp);
2435 else
2436 spoe_set_var(ctx, scope, str, sz, &smp);
2437
2438 ret = (p - *buf);
2439 *buf = p;
2440 return ret;
2441 skip:
2442 return 0;
2443 }
2444
2445 static inline int
spoe_decode_action_unset_var(struct stream * s,struct spoe_context * ctx,char ** buf,char * end,int dir)2446 spoe_decode_action_unset_var(struct stream *s, struct spoe_context *ctx,
2447 char **buf, char *end, int dir)
2448 {
2449 char *str, *scope, *p = *buf;
2450 struct sample smp;
2451 uint64_t sz;
2452 int ret;
2453
2454 if (p + 2 >= end)
2455 goto skip;
2456
2457 /* UNSET-VAR requires 2 arguments */
2458 if (*p++ != 2)
2459 goto skip;
2460
2461 switch (*p++) {
2462 case SPOE_SCOPE_PROC: scope = "proc"; break;
2463 case SPOE_SCOPE_SESS: scope = "sess"; break;
2464 case SPOE_SCOPE_TXN : scope = "txn"; break;
2465 case SPOE_SCOPE_REQ : scope = "req"; break;
2466 case SPOE_SCOPE_RES : scope = "res"; break;
2467 default: goto skip;
2468 }
2469
2470 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
2471 goto skip;
2472 memset(&smp, 0, sizeof(smp));
2473 smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2474
2475 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2476 " - unset-var '%s.%s.%.*s'\n",
2477 (int)now.tv_sec, (int)now.tv_usec,
2478 ((struct spoe_config *)FLT_CONF(ctx->filter))->agent->id,
2479 __FUNCTION__, s, scope,
2480 ((struct spoe_config *)FLT_CONF(ctx->filter))->agent->var_pfx,
2481 (int)sz, str);
2482
2483 spoe_unset_var(ctx, scope, str, sz, &smp);
2484
2485 ret = (p - *buf);
2486 *buf = p;
2487 return ret;
2488 skip:
2489 return 0;
2490 }
2491
2492 /* Process SPOE actions for a specific event. It returns 1 on success. If an
2493 * error occurred, 0 is returned. */
2494 static int
spoe_process_actions(struct stream * s,struct spoe_context * ctx,int dir)2495 spoe_process_actions(struct stream *s, struct spoe_context *ctx, int dir)
2496 {
2497 char *p, *end;
2498 int ret;
2499
2500 p = b_head(&ctx->buffer);
2501 end = p + b_data(&ctx->buffer);
2502
2503 while (p < end) {
2504 enum spoe_action_type type;
2505
2506 type = *p++;
2507 switch (type) {
2508 case SPOE_ACT_T_SET_VAR:
2509 ret = spoe_decode_action_set_var(s, ctx, &p, end, dir);
2510 if (!ret)
2511 goto skip;
2512 break;
2513
2514 case SPOE_ACT_T_UNSET_VAR:
2515 ret = spoe_decode_action_unset_var(s, ctx, &p, end, dir);
2516 if (!ret)
2517 goto skip;
2518 break;
2519
2520 default:
2521 goto skip;
2522 }
2523 }
2524
2525 return 1;
2526 skip:
2527 return 0;
2528 }
2529
2530 /***************************************************************************
2531 * Functions that process SPOE events
2532 **************************************************************************/
2533 static void
spoe_update_stats(struct stream * s,struct spoe_agent * agent,struct spoe_context * ctx,int dir)2534 spoe_update_stats(struct stream *s, struct spoe_agent *agent,
2535 struct spoe_context *ctx, int dir)
2536 {
2537 if (!tv_iszero(&ctx->stats.tv_start)) {
2538 spoe_update_stat_time(&ctx->stats.tv_start, &ctx->stats.t_process);
2539 ctx->stats.t_total += ctx->stats.t_process;
2540 tv_zero(&ctx->stats.tv_request);
2541 tv_zero(&ctx->stats.tv_queue);
2542 tv_zero(&ctx->stats.tv_wait);
2543 tv_zero(&ctx->stats.tv_response);
2544 }
2545
2546 if (agent->var_t_process) {
2547 struct sample smp;
2548
2549 memset(&smp, 0, sizeof(smp));
2550 smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2551 smp.data.u.sint = ctx->stats.t_process;
2552 smp.data.type = SMP_T_SINT;
2553
2554 spoe_set_var(ctx, "txn", agent->var_t_process,
2555 strlen(agent->var_t_process), &smp);
2556 }
2557
2558 if (agent->var_t_total) {
2559 struct sample smp;
2560
2561 memset(&smp, 0, sizeof(smp));
2562 smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2563 smp.data.u.sint = ctx->stats.t_total;
2564 smp.data.type = SMP_T_SINT;
2565
2566 spoe_set_var(ctx, "txn", agent->var_t_total,
2567 strlen(agent->var_t_total), &smp);
2568 }
2569 }
2570
2571 static void
spoe_handle_processing_error(struct stream * s,struct spoe_agent * agent,struct spoe_context * ctx,int dir)2572 spoe_handle_processing_error(struct stream *s, struct spoe_agent *agent,
2573 struct spoe_context *ctx, int dir)
2574 {
2575 if (agent->eps_max > 0)
2576 update_freq_ctr(&agent->rt[tid].err_per_sec, 1);
2577
2578 if (agent->var_on_error) {
2579 struct sample smp;
2580
2581 memset(&smp, 0, sizeof(smp));
2582 smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2583 smp.data.u.sint = ctx->status_code;
2584 smp.data.type = SMP_T_BOOL;
2585
2586 spoe_set_var(ctx, "txn", agent->var_on_error,
2587 strlen(agent->var_on_error), &smp);
2588 }
2589
2590 ctx->state = ((agent->flags & SPOE_FL_CONT_ON_ERR)
2591 ? SPOE_CTX_ST_READY
2592 : SPOE_CTX_ST_NONE);
2593 }
2594
2595 static inline int
spoe_start_processing(struct spoe_agent * agent,struct spoe_context * ctx,int dir)2596 spoe_start_processing(struct spoe_agent *agent, struct spoe_context *ctx, int dir)
2597 {
2598 /* If a process is already started for this SPOE context, retry
2599 * later. */
2600 if (ctx->flags & SPOE_CTX_FL_PROCESS)
2601 return 0;
2602
2603 agent->rt[tid].processing++;
2604 ctx->stats.tv_start = now;
2605 ctx->stats.tv_request = now;
2606 ctx->stats.t_request = -1;
2607 ctx->stats.t_queue = -1;
2608 ctx->stats.t_waiting = -1;
2609 ctx->stats.t_response = -1;
2610 ctx->stats.t_process = -1;
2611
2612 ctx->status_code = 0;
2613
2614 /* Set the right flag to prevent request and response processing
2615 * in same time. */
2616 ctx->flags |= ((dir == SMP_OPT_DIR_REQ)
2617 ? SPOE_CTX_FL_REQ_PROCESS
2618 : SPOE_CTX_FL_RSP_PROCESS);
2619 return 1;
2620 }
2621
2622 static inline void
spoe_stop_processing(struct spoe_agent * agent,struct spoe_context * ctx)2623 spoe_stop_processing(struct spoe_agent *agent, struct spoe_context *ctx)
2624 {
2625 struct spoe_appctx *sa = ctx->spoe_appctx;
2626
2627 if (!(ctx->flags & SPOE_CTX_FL_PROCESS))
2628 return;
2629 _HA_ATOMIC_ADD(&agent->counters.nb_processed, 1);
2630 if (sa) {
2631 if (sa->frag_ctx.ctx == ctx) {
2632 sa->frag_ctx.ctx = NULL;
2633 spoe_wakeup_appctx(sa->owner);
2634 }
2635 else
2636 sa->cur_fpa--;
2637 }
2638
2639 /* Reset the flag to allow next processing */
2640 agent->rt[tid].processing--;
2641 ctx->flags &= ~(SPOE_CTX_FL_PROCESS|SPOE_CTX_FL_FRAGMENTED);
2642
2643 /* Reset processing timer */
2644 ctx->process_exp = TICK_ETERNITY;
2645
2646 spoe_release_buffer(&ctx->buffer, &ctx->buffer_wait);
2647
2648 ctx->spoe_appctx = NULL;
2649 ctx->frag_ctx.curmsg = NULL;
2650 ctx->frag_ctx.curarg = NULL;
2651 ctx->frag_ctx.curoff = 0;
2652 ctx->frag_ctx.flags = 0;
2653
2654 if (!LIST_ISEMPTY(&ctx->list)) {
2655 if (ctx->state == SPOE_CTX_ST_SENDING_MSGS)
2656 _HA_ATOMIC_SUB(&agent->counters.nb_sending, 1);
2657 else
2658 _HA_ATOMIC_SUB(&agent->counters.nb_waiting, 1);
2659
2660 LIST_DEL(&ctx->list);
2661 LIST_INIT(&ctx->list);
2662 }
2663 }
2664
2665 /* Process a list of SPOE messages. First, this functions will process messages
2666 * and send them to an agent in a NOTIFY frame. Then, it will wait a ACK frame
2667 * to process corresponding actions. During all the processing, it returns 0
2668 * and it returns 1 when the processing is finished. If an error occurred, -1
2669 * is returned. */
2670 static int
spoe_process_messages(struct stream * s,struct spoe_context * ctx,struct list * messages,int dir,int type)2671 spoe_process_messages(struct stream *s, struct spoe_context *ctx,
2672 struct list *messages, int dir, int type)
2673 {
2674 struct spoe_config *conf = FLT_CONF(ctx->filter);
2675 struct spoe_agent *agent = conf->agent;
2676 int ret = 1;
2677
2678 if (ctx->state == SPOE_CTX_ST_ERROR)
2679 goto end;
2680
2681 if (tick_is_expired(ctx->process_exp, now_ms) && ctx->state != SPOE_CTX_ST_DONE) {
2682 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2683 " - failed to process messages: timeout\n",
2684 (int)now.tv_sec, (int)now.tv_usec,
2685 agent->id, __FUNCTION__, s);
2686 ctx->status_code = SPOE_CTX_ERR_TOUT;
2687 goto end;
2688 }
2689
2690 if (ctx->state == SPOE_CTX_ST_READY) {
2691 if (agent->eps_max > 0) {
2692 if (!freq_ctr_remain(&agent->rt[tid].err_per_sec, agent->eps_max, 0)) {
2693 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2694 " - skip processing of messages: max EPS reached\n",
2695 (int)now.tv_sec, (int)now.tv_usec,
2696 agent->id, __FUNCTION__, s);
2697 goto skip;
2698 }
2699 }
2700
2701 if (!tick_isset(ctx->process_exp)) {
2702 ctx->process_exp = tick_add_ifset(now_ms, agent->timeout.processing);
2703 s->task->expire = tick_first((tick_is_expired(s->task->expire, now_ms) ? 0 : s->task->expire),
2704 ctx->process_exp);
2705 }
2706 ret = spoe_start_processing(agent, ctx, dir);
2707 if (!ret)
2708 goto out;
2709
2710 ctx->state = SPOE_CTX_ST_ENCODING_MSGS;
2711 /* fall through */
2712 }
2713
2714 if (ctx->state == SPOE_CTX_ST_ENCODING_MSGS) {
2715 if (tv_iszero(&ctx->stats.tv_request))
2716 ctx->stats.tv_request = now;
2717 if (!spoe_acquire_buffer(&ctx->buffer, &ctx->buffer_wait))
2718 goto out;
2719 ret = spoe_encode_messages(s, ctx, messages, dir, type);
2720 if (ret < 0)
2721 goto end;
2722 if (!ret)
2723 goto skip;
2724 if (spoe_queue_context(ctx) < 0)
2725 goto end;
2726 ctx->state = SPOE_CTX_ST_SENDING_MSGS;
2727 }
2728
2729 if (ctx->state == SPOE_CTX_ST_SENDING_MSGS) {
2730 if (ctx->spoe_appctx)
2731 spoe_wakeup_appctx(ctx->spoe_appctx->owner);
2732 ret = 0;
2733 goto out;
2734 }
2735
2736 if (ctx->state == SPOE_CTX_ST_WAITING_ACK) {
2737 ret = 0;
2738 goto out;
2739 }
2740
2741 if (ctx->state == SPOE_CTX_ST_DONE) {
2742 spoe_process_actions(s, ctx, dir);
2743 ret = 1;
2744 ctx->frame_id++;
2745 ctx->state = SPOE_CTX_ST_READY;
2746 spoe_update_stat_time(&ctx->stats.tv_response, &ctx->stats.t_response);
2747 goto end;
2748 }
2749
2750 out:
2751 return ret;
2752
2753 skip:
2754 tv_zero(&ctx->stats.tv_start);
2755 ctx->state = SPOE_CTX_ST_READY;
2756 spoe_stop_processing(agent, ctx);
2757 return 1;
2758
2759 end:
2760 spoe_update_stats(s, agent, ctx, dir);
2761 spoe_stop_processing(agent, ctx);
2762 if (ctx->status_code) {
2763 _HA_ATOMIC_ADD(&agent->counters.nb_errors, 1);
2764 spoe_handle_processing_error(s, agent, ctx, dir);
2765 ret = 1;
2766 }
2767 return ret;
2768 }
2769
2770 /* Process a SPOE group, ie the list of messages attached to the group <grp>.
2771 * See spoe_process_message for details. */
2772 static int
spoe_process_group(struct stream * s,struct spoe_context * ctx,struct spoe_group * group,int dir)2773 spoe_process_group(struct stream *s, struct spoe_context *ctx,
2774 struct spoe_group *group, int dir)
2775 {
2776 struct spoe_config *conf = FLT_CONF(ctx->filter);
2777 struct spoe_agent *agent = conf->agent;
2778 int ret;
2779
2780 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2781 " - ctx-state=%s - Process messages for group=%s\n",
2782 (int)now.tv_sec, (int)now.tv_usec, agent->id,
2783 __FUNCTION__, s, spoe_ctx_state_str[ctx->state],
2784 group->id);
2785
2786 if (LIST_ISEMPTY(&group->messages))
2787 return 1;
2788
2789 ret = spoe_process_messages(s, ctx, &group->messages, dir, SPOE_MSGS_BY_GROUP);
2790 if (ret && ctx->stats.t_process != -1) {
2791 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2792 " - <GROUP:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu %u/%u\n",
2793 (int)now.tv_sec, (int)now.tv_usec, agent->id,
2794 __FUNCTION__, s, group->id, s->uniq_id, ctx->status_code,
2795 ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting,
2796 ctx->stats.t_response, ctx->stats.t_process,
2797 agent->counters.idles, agent->counters.applets,
2798 agent->counters.nb_sending, agent->counters.nb_waiting,
2799 agent->counters.nb_errors, agent->counters.nb_processed,
2800 agent->rt[tid].processing, read_freq_ctr(&agent->rt[tid].processing_per_sec));
2801 if (ctx->status_code || !(conf->agent_fe.options2 & PR_O2_NOLOGNORM))
2802 send_log(&conf->agent_fe, (!ctx->status_code ? LOG_NOTICE : LOG_WARNING),
2803 "SPOE: [%s] <GROUP:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu\n",
2804 agent->id, group->id, s->uniq_id, ctx->status_code,
2805 ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting,
2806 ctx->stats.t_response, ctx->stats.t_process,
2807 agent->counters.idles, agent->counters.applets,
2808 agent->counters.nb_sending, agent->counters.nb_waiting,
2809 agent->counters.nb_errors, agent->counters.nb_processed);
2810 }
2811 return ret;
2812 }
2813
2814 /* Process a SPOE event, ie the list of messages attached to the event <ev>.
2815 * See spoe_process_message for details. */
2816 static int
spoe_process_event(struct stream * s,struct spoe_context * ctx,enum spoe_event ev)2817 spoe_process_event(struct stream *s, struct spoe_context *ctx,
2818 enum spoe_event ev)
2819 {
2820 struct spoe_config *conf = FLT_CONF(ctx->filter);
2821 struct spoe_agent *agent = conf->agent;
2822 int dir, ret;
2823
2824 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2825 " - ctx-state=%s - Process messages for event=%s\n",
2826 (int)now.tv_sec, (int)now.tv_usec, agent->id,
2827 __FUNCTION__, s, spoe_ctx_state_str[ctx->state],
2828 spoe_event_str[ev]);
2829
2830 dir = ((ev < SPOE_EV_ON_SERVER_SESS) ? SMP_OPT_DIR_REQ : SMP_OPT_DIR_RES);
2831
2832 if (LIST_ISEMPTY(&(ctx->events[ev])))
2833 return 1;
2834
2835 ret = spoe_process_messages(s, ctx, &(ctx->events[ev]), dir, SPOE_MSGS_BY_EVENT);
2836 if (ret && ctx->stats.t_process != -1) {
2837 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2838 " - <EVENT:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu %u/%u\n",
2839 (int)now.tv_sec, (int)now.tv_usec, agent->id,
2840 __FUNCTION__, s, spoe_event_str[ev], s->uniq_id, ctx->status_code,
2841 ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting,
2842 ctx->stats.t_response, ctx->stats.t_process,
2843 agent->counters.idles, agent->counters.applets,
2844 agent->counters.nb_sending, agent->counters.nb_waiting,
2845 agent->counters.nb_errors, agent->counters.nb_processed,
2846 agent->rt[tid].processing, read_freq_ctr(&agent->rt[tid].processing_per_sec));
2847 if (ctx->status_code || !(conf->agent_fe.options2 & PR_O2_NOLOGNORM))
2848 send_log(&conf->agent_fe, (!ctx->status_code ? LOG_NOTICE : LOG_WARNING),
2849 "SPOE: [%s] <EVENT:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu\n",
2850 agent->id, spoe_event_str[ev], s->uniq_id, ctx->status_code,
2851 ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting,
2852 ctx->stats.t_response, ctx->stats.t_process,
2853 agent->counters.idles, agent->counters.applets,
2854 agent->counters.nb_sending, agent->counters.nb_waiting,
2855 agent->counters.nb_errors, agent->counters.nb_processed);
2856 }
2857 return ret;
2858 }
2859
2860 /***************************************************************************
2861 * Functions that create/destroy SPOE contexts
2862 **************************************************************************/
2863 static int
spoe_acquire_buffer(struct buffer * buf,struct buffer_wait * buffer_wait)2864 spoe_acquire_buffer(struct buffer *buf, struct buffer_wait *buffer_wait)
2865 {
2866 if (buf->size)
2867 return 1;
2868
2869 if (!LIST_ISEMPTY(&buffer_wait->list)) {
2870 HA_SPIN_LOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2871 LIST_DEL(&buffer_wait->list);
2872 LIST_INIT(&buffer_wait->list);
2873 HA_SPIN_UNLOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2874 }
2875
2876 if (b_alloc_margin(buf, global.tune.reserved_bufs))
2877 return 1;
2878
2879 HA_SPIN_LOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2880 LIST_ADDQ(&buffer_wq, &buffer_wait->list);
2881 HA_SPIN_UNLOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2882 return 0;
2883 }
2884
2885 static void
spoe_release_buffer(struct buffer * buf,struct buffer_wait * buffer_wait)2886 spoe_release_buffer(struct buffer *buf, struct buffer_wait *buffer_wait)
2887 {
2888 if (!LIST_ISEMPTY(&buffer_wait->list)) {
2889 HA_SPIN_LOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2890 LIST_DEL(&buffer_wait->list);
2891 LIST_INIT(&buffer_wait->list);
2892 HA_SPIN_UNLOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2893 }
2894
2895 /* Release the buffer if needed */
2896 if (buf->size) {
2897 b_free(buf);
2898 offer_buffers(buffer_wait->target, tasks_run_queue);
2899 }
2900 }
2901
2902 static int
spoe_wakeup_context(struct spoe_context * ctx)2903 spoe_wakeup_context(struct spoe_context *ctx)
2904 {
2905 task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
2906 return 1;
2907 }
2908
2909 static struct spoe_context *
spoe_create_context(struct stream * s,struct filter * filter)2910 spoe_create_context(struct stream *s, struct filter *filter)
2911 {
2912 struct spoe_config *conf = FLT_CONF(filter);
2913 struct spoe_context *ctx;
2914
2915 ctx = pool_alloc_dirty(pool_head_spoe_ctx);
2916 if (ctx == NULL) {
2917 return NULL;
2918 }
2919 memset(ctx, 0, sizeof(*ctx));
2920 ctx->filter = filter;
2921 ctx->state = SPOE_CTX_ST_NONE;
2922 ctx->status_code = SPOE_CTX_ERR_NONE;
2923 ctx->flags = 0;
2924 ctx->events = conf->agent->events;
2925 ctx->groups = &conf->agent->groups;
2926 ctx->buffer = BUF_NULL;
2927 LIST_INIT(&ctx->buffer_wait.list);
2928 ctx->buffer_wait.target = ctx;
2929 ctx->buffer_wait.wakeup_cb = (int (*)(void *))spoe_wakeup_context;
2930 LIST_INIT(&ctx->list);
2931
2932 ctx->stream_id = 0;
2933 ctx->frame_id = 1;
2934 ctx->process_exp = TICK_ETERNITY;
2935
2936 tv_zero(&ctx->stats.tv_start);
2937 tv_zero(&ctx->stats.tv_request);
2938 tv_zero(&ctx->stats.tv_queue);
2939 tv_zero(&ctx->stats.tv_wait);
2940 tv_zero(&ctx->stats.tv_response);
2941 ctx->stats.t_request = -1;
2942 ctx->stats.t_queue = -1;
2943 ctx->stats.t_waiting = -1;
2944 ctx->stats.t_response = -1;
2945 ctx->stats.t_process = -1;
2946 ctx->stats.t_total = 0;
2947
2948 ctx->strm = s;
2949 ctx->state = SPOE_CTX_ST_READY;
2950 filter->ctx = ctx;
2951
2952 return ctx;
2953 }
2954
2955 static void
spoe_destroy_context(struct filter * filter)2956 spoe_destroy_context(struct filter *filter)
2957 {
2958 struct spoe_config *conf = FLT_CONF(filter);
2959 struct spoe_context *ctx = filter->ctx;
2960
2961 if (!ctx)
2962 return;
2963
2964 spoe_stop_processing(conf->agent, ctx);
2965 pool_free(pool_head_spoe_ctx, ctx);
2966 filter->ctx = NULL;
2967 }
2968
2969 static void
spoe_reset_context(struct spoe_context * ctx)2970 spoe_reset_context(struct spoe_context *ctx)
2971 {
2972 ctx->state = SPOE_CTX_ST_READY;
2973 ctx->flags &= ~(SPOE_CTX_FL_PROCESS|SPOE_CTX_FL_FRAGMENTED);
2974
2975 tv_zero(&ctx->stats.tv_start);
2976 tv_zero(&ctx->stats.tv_request);
2977 tv_zero(&ctx->stats.tv_queue);
2978 tv_zero(&ctx->stats.tv_wait);
2979 tv_zero(&ctx->stats.tv_response);
2980 ctx->stats.t_request = -1;
2981 ctx->stats.t_queue = -1;
2982 ctx->stats.t_waiting = -1;
2983 ctx->stats.t_response = -1;
2984 ctx->stats.t_process = -1;
2985 ctx->stats.t_total = 0;
2986 }
2987
2988
2989 /***************************************************************************
2990 * Hooks that manage the filter lifecycle (init/check/deinit)
2991 **************************************************************************/
2992 /* Signal handler: Do a soft stop, wakeup SPOE applet */
2993 static void
spoe_sig_stop(struct sig_handler * sh)2994 spoe_sig_stop(struct sig_handler *sh)
2995 {
2996 struct proxy *p;
2997
2998 p = proxies_list;
2999 while (p) {
3000 struct flt_conf *fconf;
3001
3002 list_for_each_entry(fconf, &p->filter_configs, list) {
3003 struct spoe_config *conf;
3004 struct spoe_agent *agent;
3005 struct spoe_appctx *spoe_appctx;
3006 int i;
3007
3008 if (fconf->id != spoe_filter_id)
3009 continue;
3010
3011 conf = fconf->conf;
3012 agent = conf->agent;
3013
3014 for (i = 0; i < global.nbthread; ++i) {
3015 HA_SPIN_LOCK(SPOE_APPLET_LOCK, &agent->rt[i].lock);
3016 list_for_each_entry(spoe_appctx, &agent->rt[i].applets, list)
3017 spoe_wakeup_appctx(spoe_appctx->owner);
3018 HA_SPIN_UNLOCK(SPOE_APPLET_LOCK, &agent->rt[i].lock);
3019 }
3020 }
3021 p = p->next;
3022 }
3023 }
3024
3025
3026 /* Initialize the SPOE filter. Returns -1 on error, else 0. */
3027 static int
spoe_init(struct proxy * px,struct flt_conf * fconf)3028 spoe_init(struct proxy *px, struct flt_conf *fconf)
3029 {
3030 struct spoe_config *conf = fconf->conf;
3031
3032 /* conf->agent_fe was already initialized during the config
3033 * parsing. Finish initialization. */
3034 conf->agent_fe.last_change = now.tv_sec;
3035 conf->agent_fe.cap = PR_CAP_FE;
3036 conf->agent_fe.mode = PR_MODE_TCP;
3037 conf->agent_fe.maxconn = 0;
3038 conf->agent_fe.options2 |= PR_O2_INDEPSTR;
3039 conf->agent_fe.conn_retries = CONN_RETRIES;
3040 conf->agent_fe.accept = frontend_accept;
3041 conf->agent_fe.srv = NULL;
3042 conf->agent_fe.timeout.client = TICK_ETERNITY;
3043 conf->agent_fe.default_target = &spoe_applet.obj_type;
3044 conf->agent_fe.fe_req_ana = AN_REQ_SWITCHING_RULES;
3045
3046 if (!sighandler_registered) {
3047 signal_register_fct(0, spoe_sig_stop, 0);
3048 sighandler_registered = 1;
3049 }
3050
3051 fconf->flags |= FLT_CFG_FL_HTX;
3052 return 0;
3053 }
3054
3055 /* Free ressources allocated by the SPOE filter. */
3056 static void
spoe_deinit(struct proxy * px,struct flt_conf * fconf)3057 spoe_deinit(struct proxy *px, struct flt_conf *fconf)
3058 {
3059 struct spoe_config *conf = fconf->conf;
3060
3061 if (conf) {
3062 struct spoe_agent *agent = conf->agent;
3063
3064 spoe_release_agent(agent);
3065 free(conf->id);
3066 free(conf);
3067 }
3068 fconf->conf = NULL;
3069 }
3070
3071 /* Check configuration of a SPOE filter for a specified proxy.
3072 * Return 1 on error, else 0. */
3073 static int
spoe_check(struct proxy * px,struct flt_conf * fconf)3074 spoe_check(struct proxy *px, struct flt_conf *fconf)
3075 {
3076 struct flt_conf *f;
3077 struct spoe_config *conf = fconf->conf;
3078 struct proxy *target;
3079 int i;
3080
3081 /* Check all SPOE filters for proxy <px> to be sure all SPOE agent names
3082 * are uniq */
3083 list_for_each_entry(f, &px->filter_configs, list) {
3084 struct spoe_config *c = f->conf;
3085
3086 /* This is not an SPOE filter */
3087 if (f->id != spoe_filter_id)
3088 continue;
3089 /* This is the current SPOE filter */
3090 if (f == fconf)
3091 continue;
3092
3093 /* Check engine Id. It should be uniq */
3094 if (!strcmp(conf->id, c->id)) {
3095 ha_alert("Proxy %s : duplicated name for SPOE engine '%s'.\n",
3096 px->id, conf->id);
3097 return 1;
3098 }
3099 }
3100
3101 target = proxy_be_by_name(conf->agent->b.name);
3102 if (target == NULL) {
3103 ha_alert("Proxy %s : unknown backend '%s' used by SPOE agent '%s'"
3104 " declared at %s:%d.\n",
3105 px->id, conf->agent->b.name, conf->agent->id,
3106 conf->agent->conf.file, conf->agent->conf.line);
3107 return 1;
3108 }
3109 if (target->mode != PR_MODE_TCP) {
3110 ha_alert("Proxy %s : backend '%s' used by SPOE agent '%s' declared"
3111 " at %s:%d does not support HTTP mode.\n",
3112 px->id, target->id, conf->agent->id,
3113 conf->agent->conf.file, conf->agent->conf.line);
3114 return 1;
3115 }
3116
3117 if (px->bind_proc & ~target->bind_proc) {
3118 ha_alert("Proxy %s : backend '%s' used by SPOE agent '%s' declared"
3119 " at %s:%d does not cover all of its processes.\n",
3120 px->id, target->id, conf->agent->id,
3121 conf->agent->conf.file, conf->agent->conf.line);
3122 return 1;
3123 }
3124
3125 if ((conf->agent->rt = calloc(global.nbthread, sizeof(*conf->agent->rt))) == NULL) {
3126 ha_alert("Proxy %s : out of memory initializing SPOE agent '%s' declared at %s:%d.\n",
3127 px->id, conf->agent->id, conf->agent->conf.file, conf->agent->conf.line);
3128 return 1;
3129 }
3130 for (i = 0; i < global.nbthread; ++i) {
3131 conf->agent->rt[i].engine_id = NULL;
3132 conf->agent->rt[i].frame_size = conf->agent->max_frame_size;
3133 conf->agent->rt[i].processing = 0;
3134 LIST_INIT(&conf->agent->rt[i].applets);
3135 LIST_INIT(&conf->agent->rt[i].sending_queue);
3136 LIST_INIT(&conf->agent->rt[i].waiting_queue);
3137 HA_SPIN_INIT(&conf->agent->rt[i].lock);
3138 }
3139
3140 free(conf->agent->b.name);
3141 conf->agent->b.name = NULL;
3142 conf->agent->b.be = target;
3143 return 0;
3144 }
3145
3146 /* Initializes the SPOE filter for a proxy for a specific thread.
3147 * Returns a negative value if an error occurs. */
3148 static int
spoe_init_per_thread(struct proxy * p,struct flt_conf * fconf)3149 spoe_init_per_thread(struct proxy *p, struct flt_conf *fconf)
3150 {
3151 struct spoe_config *conf = fconf->conf;
3152 struct spoe_agent *agent = conf->agent;
3153
3154 agent->rt[tid].engine_id = generate_pseudo_uuid();
3155 if (agent->rt[tid].engine_id == NULL)
3156 return -1;
3157 return 0;
3158 }
3159
3160 /**************************************************************************
3161 * Hooks attached to a stream
3162 *************************************************************************/
3163 /* Called when a filter instance is created and attach to a stream. It creates
3164 * the context that will be used to process this stream. */
3165 static int
spoe_start(struct stream * s,struct filter * filter)3166 spoe_start(struct stream *s, struct filter *filter)
3167 {
3168 struct spoe_config *conf = FLT_CONF(filter);
3169 struct spoe_agent *agent = conf->agent;
3170 struct spoe_context *ctx;
3171
3172 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p\n",
3173 (int)now.tv_sec, (int)now.tv_usec, agent->id,
3174 __FUNCTION__, s);
3175
3176 if ((ctx = spoe_create_context(s, filter)) == NULL) {
3177 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
3178 " - failed to create SPOE context\n",
3179 (int)now.tv_sec, (int)now.tv_usec, agent->id,
3180 __FUNCTION__, s);
3181 send_log(&conf->agent_fe, LOG_EMERG,
3182 "SPOE: [%s] failed to create SPOE context\n",
3183 agent->id);
3184 return 0;
3185 }
3186
3187 if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_TCP_REQ_FE]))
3188 filter->pre_analyzers |= AN_REQ_INSPECT_FE;
3189
3190 if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_TCP_REQ_BE]))
3191 filter->pre_analyzers |= AN_REQ_INSPECT_BE;
3192
3193 if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_TCP_RSP]))
3194 filter->pre_analyzers |= AN_RES_INSPECT;
3195
3196 if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_HTTP_REQ_FE]))
3197 filter->pre_analyzers |= AN_REQ_HTTP_PROCESS_FE;
3198
3199 if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_HTTP_REQ_BE]))
3200 filter->pre_analyzers |= AN_REQ_HTTP_PROCESS_BE;
3201
3202 if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_HTTP_RSP]))
3203 filter->pre_analyzers |= AN_RES_HTTP_PROCESS_FE;
3204
3205 return 1;
3206 }
3207
3208 /* Called when a filter instance is detached from a stream. It release the
3209 * attached SPOE context. */
3210 static void
spoe_stop(struct stream * s,struct filter * filter)3211 spoe_stop(struct stream *s, struct filter *filter)
3212 {
3213 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p\n",
3214 (int)now.tv_sec, (int)now.tv_usec,
3215 ((struct spoe_config *)FLT_CONF(filter))->agent->id,
3216 __FUNCTION__, s);
3217 spoe_destroy_context(filter);
3218 }
3219
3220
3221 /*
3222 * Called when the stream is woken up because of expired timer.
3223 */
3224 static void
spoe_check_timeouts(struct stream * s,struct filter * filter)3225 spoe_check_timeouts(struct stream *s, struct filter *filter)
3226 {
3227 struct spoe_context *ctx = filter->ctx;
3228
3229 if (tick_is_expired(ctx->process_exp, now_ms))
3230 s->pending_events |= TASK_WOKEN_MSG;
3231 }
3232
3233 /* Called when we are ready to filter data on a channel */
3234 static int
spoe_start_analyze(struct stream * s,struct filter * filter,struct channel * chn)3235 spoe_start_analyze(struct stream *s, struct filter *filter, struct channel *chn)
3236 {
3237 struct spoe_context *ctx = filter->ctx;
3238 int ret = 1;
3239
3240 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p - ctx-state=%s"
3241 " - ctx-flags=0x%08x\n",
3242 (int)now.tv_sec, (int)now.tv_usec,
3243 ((struct spoe_config *)FLT_CONF(filter))->agent->id,
3244 __FUNCTION__, s, spoe_ctx_state_str[ctx->state], ctx->flags);
3245
3246 if (ctx->state == SPOE_CTX_ST_NONE)
3247 goto out;
3248
3249 if (!(chn->flags & CF_ISRESP)) {
3250 if (filter->pre_analyzers & AN_REQ_INSPECT_FE)
3251 chn->analysers |= AN_REQ_INSPECT_FE;
3252 if (filter->pre_analyzers & AN_REQ_INSPECT_BE)
3253 chn->analysers |= AN_REQ_INSPECT_BE;
3254
3255 if (ctx->flags & SPOE_CTX_FL_CLI_CONNECTED)
3256 goto out;
3257
3258 ctx->stream_id = s->uniq_id;
3259 ret = spoe_process_event(s, ctx, SPOE_EV_ON_CLIENT_SESS);
3260 if (!ret)
3261 goto out;
3262 ctx->flags |= SPOE_CTX_FL_CLI_CONNECTED;
3263 }
3264 else {
3265 if (filter->pre_analyzers & AN_RES_INSPECT)
3266 chn->analysers |= AN_RES_INSPECT;
3267
3268 if (ctx->flags & SPOE_CTX_FL_SRV_CONNECTED)
3269 goto out;
3270
3271 ret = spoe_process_event(s, ctx, SPOE_EV_ON_SERVER_SESS);
3272 if (!ret) {
3273 channel_dont_read(chn);
3274 channel_dont_close(chn);
3275 goto out;
3276 }
3277 ctx->flags |= SPOE_CTX_FL_SRV_CONNECTED;
3278 }
3279
3280 out:
3281 return ret;
3282 }
3283
3284 /* Called before a processing happens on a given channel */
3285 static int
spoe_chn_pre_analyze(struct stream * s,struct filter * filter,struct channel * chn,unsigned an_bit)3286 spoe_chn_pre_analyze(struct stream *s, struct filter *filter,
3287 struct channel *chn, unsigned an_bit)
3288 {
3289 struct spoe_context *ctx = filter->ctx;
3290 int ret = 1;
3291
3292 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p - ctx-state=%s"
3293 " - ctx-flags=0x%08x - ana=0x%08x\n",
3294 (int)now.tv_sec, (int)now.tv_usec,
3295 ((struct spoe_config *)FLT_CONF(filter))->agent->id,
3296 __FUNCTION__, s, spoe_ctx_state_str[ctx->state],
3297 ctx->flags, an_bit);
3298
3299 if (ctx->state == SPOE_CTX_ST_NONE)
3300 goto out;
3301
3302 switch (an_bit) {
3303 case AN_REQ_INSPECT_FE:
3304 ret = spoe_process_event(s, ctx, SPOE_EV_ON_TCP_REQ_FE);
3305 break;
3306 case AN_REQ_INSPECT_BE:
3307 ret = spoe_process_event(s, ctx, SPOE_EV_ON_TCP_REQ_BE);
3308 break;
3309 case AN_RES_INSPECT:
3310 ret = spoe_process_event(s, ctx, SPOE_EV_ON_TCP_RSP);
3311 break;
3312 case AN_REQ_HTTP_PROCESS_FE:
3313 ret = spoe_process_event(s, ctx, SPOE_EV_ON_HTTP_REQ_FE);
3314 break;
3315 case AN_REQ_HTTP_PROCESS_BE:
3316 ret = spoe_process_event(s, ctx, SPOE_EV_ON_HTTP_REQ_BE);
3317 break;
3318 case AN_RES_HTTP_PROCESS_FE:
3319 ret = spoe_process_event(s, ctx, SPOE_EV_ON_HTTP_RSP);
3320 break;
3321 }
3322
3323 out:
3324 if (!ret && (chn->flags & CF_ISRESP)) {
3325 channel_dont_read(chn);
3326 channel_dont_close(chn);
3327 }
3328 return ret;
3329 }
3330
3331 /* Called when the filtering on the channel ends. */
3332 static int
spoe_end_analyze(struct stream * s,struct filter * filter,struct channel * chn)3333 spoe_end_analyze(struct stream *s, struct filter *filter, struct channel *chn)
3334 {
3335 struct spoe_context *ctx = filter->ctx;
3336
3337 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p - ctx-state=%s"
3338 " - ctx-flags=0x%08x\n",
3339 (int)now.tv_sec, (int)now.tv_usec,
3340 ((struct spoe_config *)FLT_CONF(filter))->agent->id,
3341 __FUNCTION__, s, spoe_ctx_state_str[ctx->state], ctx->flags);
3342
3343 if (!(ctx->flags & SPOE_CTX_FL_PROCESS)) {
3344 spoe_reset_context(ctx);
3345 }
3346
3347 return 1;
3348 }
3349
3350 /********************************************************************
3351 * Functions that manage the filter initialization
3352 ********************************************************************/
3353 struct flt_ops spoe_ops = {
3354 /* Manage SPOE filter, called for each filter declaration */
3355 .init = spoe_init,
3356 .deinit = spoe_deinit,
3357 .check = spoe_check,
3358 .init_per_thread = spoe_init_per_thread,
3359
3360 /* Handle start/stop of SPOE */
3361 .attach = spoe_start,
3362 .detach = spoe_stop,
3363 .check_timeouts = spoe_check_timeouts,
3364
3365 /* Handle channels activity */
3366 .channel_start_analyze = spoe_start_analyze,
3367 .channel_pre_analyze = spoe_chn_pre_analyze,
3368 .channel_end_analyze = spoe_end_analyze,
3369 };
3370
3371
3372 static int
cfg_parse_spoe_agent(const char * file,int linenum,char ** args,int kwm)3373 cfg_parse_spoe_agent(const char *file, int linenum, char **args, int kwm)
3374 {
3375 const char *err;
3376 int i, err_code = 0;
3377
3378 if ((cfg_scope == NULL && curengine != NULL) ||
3379 (cfg_scope != NULL && curengine == NULL) ||
3380 (curengine != NULL && cfg_scope != NULL && strcmp(curengine, cfg_scope)))
3381 goto out;
3382
3383 if (!strcmp(args[0], "spoe-agent")) { /* new spoe-agent section */
3384 if (!*args[1]) {
3385 ha_alert("parsing [%s:%d] : missing name for spoe-agent section.\n",
3386 file, linenum);
3387 err_code |= ERR_ALERT | ERR_ABORT;
3388 goto out;
3389 }
3390 if (alertif_too_many_args(1, file, linenum, args, &err_code)) {
3391 err_code |= ERR_ABORT;
3392 goto out;
3393 }
3394
3395 err = invalid_char(args[1]);
3396 if (err) {
3397 ha_alert("parsing [%s:%d] : character '%c' is not permitted in '%s' name '%s'.\n",
3398 file, linenum, *err, args[0], args[1]);
3399 err_code |= ERR_ALERT | ERR_ABORT;
3400 goto out;
3401 }
3402
3403 if (curagent != NULL) {
3404 ha_alert("parsing [%s:%d] : another spoe-agent section previously defined.\n",
3405 file, linenum);
3406 err_code |= ERR_ALERT | ERR_ABORT;
3407 goto out;
3408 }
3409 if ((curagent = calloc(1, sizeof(*curagent))) == NULL) {
3410 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3411 err_code |= ERR_ALERT | ERR_ABORT;
3412 goto out;
3413 }
3414
3415 curagent->id = strdup(args[1]);
3416
3417 curagent->conf.file = strdup(file);
3418 curagent->conf.line = linenum;
3419
3420 curagent->timeout.hello = TICK_ETERNITY;
3421 curagent->timeout.idle = TICK_ETERNITY;
3422 curagent->timeout.processing = TICK_ETERNITY;
3423
3424 curagent->var_pfx = NULL;
3425 curagent->var_on_error = NULL;
3426 curagent->var_t_process = NULL;
3427 curagent->var_t_total = NULL;
3428 curagent->flags = (SPOE_FL_ASYNC | SPOE_FL_PIPELINING | SPOE_FL_SND_FRAGMENTATION);
3429 curagent->cps_max = 0;
3430 curagent->eps_max = 0;
3431 curagent->max_frame_size = MAX_FRAME_SIZE;
3432 curagent->max_fpa = 20;
3433
3434 for (i = 0; i < SPOE_EV_EVENTS; ++i)
3435 LIST_INIT(&curagent->events[i]);
3436 LIST_INIT(&curagent->groups);
3437 LIST_INIT(&curagent->messages);
3438 }
3439 else if (!strcmp(args[0], "use-backend")) {
3440 if (!*args[1]) {
3441 ha_alert("parsing [%s:%d] : '%s' expects a backend name.\n",
3442 file, linenum, args[0]);
3443 err_code |= ERR_ALERT | ERR_FATAL;
3444 goto out;
3445 }
3446 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3447 goto out;
3448 free(curagent->b.name);
3449 curagent->b.name = strdup(args[1]);
3450 }
3451 else if (!strcmp(args[0], "messages")) {
3452 int cur_arg = 1;
3453 while (*args[cur_arg]) {
3454 struct spoe_placeholder *ph = NULL;
3455
3456 list_for_each_entry(ph, &curmphs, list) {
3457 if (!strcmp(ph->id, args[cur_arg])) {
3458 ha_alert("parsing [%s:%d]: spoe-message '%s' already used.\n",
3459 file, linenum, args[cur_arg]);
3460 err_code |= ERR_ALERT | ERR_FATAL;
3461 goto out;
3462 }
3463 }
3464
3465 if ((ph = calloc(1, sizeof(*ph))) == NULL) {
3466 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3467 err_code |= ERR_ALERT | ERR_ABORT;
3468 goto out;
3469 }
3470 ph->id = strdup(args[cur_arg]);
3471 LIST_ADDQ(&curmphs, &ph->list);
3472 cur_arg++;
3473 }
3474 }
3475 else if (!strcmp(args[0], "groups")) {
3476 int cur_arg = 1;
3477 while (*args[cur_arg]) {
3478 struct spoe_placeholder *ph = NULL;
3479
3480 list_for_each_entry(ph, &curgphs, list) {
3481 if (!strcmp(ph->id, args[cur_arg])) {
3482 ha_alert("parsing [%s:%d]: spoe-group '%s' already used.\n",
3483 file, linenum, args[cur_arg]);
3484 err_code |= ERR_ALERT | ERR_FATAL;
3485 goto out;
3486 }
3487 }
3488
3489 if ((ph = calloc(1, sizeof(*ph))) == NULL) {
3490 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3491 err_code |= ERR_ALERT | ERR_ABORT;
3492 goto out;
3493 }
3494 ph->id = strdup(args[cur_arg]);
3495 LIST_ADDQ(&curgphs, &ph->list);
3496 cur_arg++;
3497 }
3498 }
3499 else if (!strcmp(args[0], "timeout")) {
3500 unsigned int *tv = NULL;
3501 const char *res;
3502 unsigned timeout;
3503
3504 if (!*args[1]) {
3505 ha_alert("parsing [%s:%d] : 'timeout' expects 'hello', 'idle' and 'processing'.\n",
3506 file, linenum);
3507 err_code |= ERR_ALERT | ERR_FATAL;
3508 goto out;
3509 }
3510 if (alertif_too_many_args(2, file, linenum, args, &err_code))
3511 goto out;
3512 if (!strcmp(args[1], "hello"))
3513 tv = &curagent->timeout.hello;
3514 else if (!strcmp(args[1], "idle"))
3515 tv = &curagent->timeout.idle;
3516 else if (!strcmp(args[1], "processing"))
3517 tv = &curagent->timeout.processing;
3518 else {
3519 ha_alert("parsing [%s:%d] : 'timeout' supports 'hello', 'idle' or 'processing' (got %s).\n",
3520 file, linenum, args[1]);
3521 err_code |= ERR_ALERT | ERR_FATAL;
3522 goto out;
3523 }
3524 if (!*args[2]) {
3525 ha_alert("parsing [%s:%d] : 'timeout %s' expects an integer value (in milliseconds).\n",
3526 file, linenum, args[1]);
3527 err_code |= ERR_ALERT | ERR_FATAL;
3528 goto out;
3529 }
3530 res = parse_time_err(args[2], &timeout, TIME_UNIT_MS);
3531 if (res == PARSE_TIME_OVER) {
3532 ha_alert("parsing [%s:%d]: timer overflow in argument <%s> to <%s %s>, maximum value is 2147483647 ms (~24.8 days).\n",
3533 file, linenum, args[2], args[0], args[1]);
3534 err_code |= ERR_ALERT | ERR_FATAL;
3535 goto out;
3536 }
3537 else if (res == PARSE_TIME_UNDER) {
3538 ha_alert("parsing [%s:%d]: timer underflow in argument <%s> to <%s %s>, minimum non-null value is 1 ms.\n",
3539 file, linenum, args[2], args[0], args[1]);
3540 err_code |= ERR_ALERT | ERR_FATAL;
3541 goto out;
3542 }
3543 else if (res) {
3544 ha_alert("parsing [%s:%d] : unexpected character '%c' in 'timeout %s'.\n",
3545 file, linenum, *res, args[1]);
3546 err_code |= ERR_ALERT | ERR_FATAL;
3547 goto out;
3548 }
3549 *tv = MS_TO_TICKS(timeout);
3550 }
3551 else if (!strcmp(args[0], "option")) {
3552 if (!*args[1]) {
3553 ha_alert("parsing [%s:%d]: '%s' expects an option name.\n",
3554 file, linenum, args[0]);
3555 err_code |= ERR_ALERT | ERR_FATAL;
3556 goto out;
3557 }
3558
3559 if (!strcmp(args[1], "pipelining")) {
3560 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3561 goto out;
3562 if (kwm == 1)
3563 curagent->flags &= ~SPOE_FL_PIPELINING;
3564 else
3565 curagent->flags |= SPOE_FL_PIPELINING;
3566 goto out;
3567 }
3568 else if (!strcmp(args[1], "async")) {
3569 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3570 goto out;
3571 if (kwm == 1)
3572 curagent->flags &= ~SPOE_FL_ASYNC;
3573 else
3574 curagent->flags |= SPOE_FL_ASYNC;
3575 goto out;
3576 }
3577 else if (!strcmp(args[1], "send-frag-payload")) {
3578 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3579 goto out;
3580 if (kwm == 1)
3581 curagent->flags &= ~SPOE_FL_SND_FRAGMENTATION;
3582 else
3583 curagent->flags |= SPOE_FL_SND_FRAGMENTATION;
3584 goto out;
3585 }
3586 else if (!strcmp(args[1], "dontlog-normal")) {
3587 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3588 goto out;
3589 if (kwm == 1)
3590 curpxopts2 &= ~PR_O2_NOLOGNORM;
3591 else
3592 curpxopts2 |= PR_O2_NOLOGNORM;
3593 goto out;
3594 }
3595
3596 /* Following options does not support negation */
3597 if (kwm == 1) {
3598 ha_alert("parsing [%s:%d]: negation is not supported for option '%s'.\n",
3599 file, linenum, args[1]);
3600 err_code |= ERR_ALERT | ERR_FATAL;
3601 goto out;
3602 }
3603
3604 if (!strcmp(args[1], "var-prefix")) {
3605 char *tmp;
3606
3607 if (!*args[2]) {
3608 ha_alert("parsing [%s:%d]: '%s %s' expects a value.\n",
3609 file, linenum, args[0],
3610 args[1]);
3611 err_code |= ERR_ALERT | ERR_FATAL;
3612 goto out;
3613 }
3614 if (alertif_too_many_args(2, file, linenum, args, &err_code))
3615 goto out;
3616 tmp = args[2];
3617 while (*tmp) {
3618 if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
3619 ha_alert("parsing [%s:%d]: '%s %s' only supports [a-zA-Z0-9_.] chars.\n",
3620 file, linenum, args[0], args[1]);
3621 err_code |= ERR_ALERT | ERR_FATAL;
3622 goto out;
3623 }
3624 tmp++;
3625 }
3626 curagent->var_pfx = strdup(args[2]);
3627 }
3628 else if (!strcmp(args[1], "force-set-var")) {
3629 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3630 goto out;
3631 curagent->flags |= SPOE_FL_FORCE_SET_VAR;
3632 }
3633 else if (!strcmp(args[1], "continue-on-error")) {
3634 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3635 goto out;
3636 curagent->flags |= SPOE_FL_CONT_ON_ERR;
3637 }
3638 else if (!strcmp(args[1], "set-on-error")) {
3639 char *tmp;
3640
3641 if (!*args[2]) {
3642 ha_alert("parsing [%s:%d]: '%s %s' expects a value.\n",
3643 file, linenum, args[0],
3644 args[1]);
3645 err_code |= ERR_ALERT | ERR_FATAL;
3646 goto out;
3647 }
3648 if (alertif_too_many_args(2, file, linenum, args, &err_code))
3649 goto out;
3650 tmp = args[2];
3651 while (*tmp) {
3652 if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
3653 ha_alert("parsing [%s:%d]: '%s %s' only supports [a-zA-Z0-9_.] chars.\n",
3654 file, linenum, args[0], args[1]);
3655 err_code |= ERR_ALERT | ERR_FATAL;
3656 goto out;
3657 }
3658 tmp++;
3659 }
3660 curagent->var_on_error = strdup(args[2]);
3661 }
3662 else if (!strcmp(args[1], "set-process-time")) {
3663 char *tmp;
3664
3665 if (!*args[2]) {
3666 ha_alert("parsing [%s:%d]: '%s %s' expects a value.\n",
3667 file, linenum, args[0],
3668 args[1]);
3669 err_code |= ERR_ALERT | ERR_FATAL;
3670 goto out;
3671 }
3672 if (alertif_too_many_args(2, file, linenum, args, &err_code))
3673 goto out;
3674 tmp = args[2];
3675 while (*tmp) {
3676 if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
3677 ha_alert("parsing [%s:%d]: '%s %s' only supports [a-zA-Z0-9_.] chars.\n",
3678 file, linenum, args[0], args[1]);
3679 err_code |= ERR_ALERT | ERR_FATAL;
3680 goto out;
3681 }
3682 tmp++;
3683 }
3684 curagent->var_t_process = strdup(args[2]);
3685 }
3686 else if (!strcmp(args[1], "set-total-time")) {
3687 char *tmp;
3688
3689 if (!*args[2]) {
3690 ha_alert("parsing [%s:%d]: '%s %s' expects a value.\n",
3691 file, linenum, args[0],
3692 args[1]);
3693 err_code |= ERR_ALERT | ERR_FATAL;
3694 goto out;
3695 }
3696 if (alertif_too_many_args(2, file, linenum, args, &err_code))
3697 goto out;
3698 tmp = args[2];
3699 while (*tmp) {
3700 if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
3701 ha_alert("parsing [%s:%d]: '%s %s' only supports [a-zA-Z0-9_.] chars.\n",
3702 file, linenum, args[0], args[1]);
3703 err_code |= ERR_ALERT | ERR_FATAL;
3704 goto out;
3705 }
3706 tmp++;
3707 }
3708 curagent->var_t_total = strdup(args[2]);
3709 }
3710 else {
3711 ha_alert("parsing [%s:%d]: option '%s' is not supported.\n",
3712 file, linenum, args[1]);
3713 err_code |= ERR_ALERT | ERR_FATAL;
3714 goto out;
3715 }
3716 }
3717 else if (!strcmp(args[0], "maxconnrate")) {
3718 if (!*args[1]) {
3719 ha_alert("parsing [%s:%d] : '%s' expects an integer argument.\n",
3720 file, linenum, args[0]);
3721 err_code |= ERR_ALERT | ERR_FATAL;
3722 goto out;
3723 }
3724 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3725 goto out;
3726 curagent->cps_max = atol(args[1]);
3727 }
3728 else if (!strcmp(args[0], "maxerrrate")) {
3729 if (!*args[1]) {
3730 ha_alert("parsing [%s:%d] : '%s' expects an integer argument.\n",
3731 file, linenum, args[0]);
3732 err_code |= ERR_ALERT | ERR_FATAL;
3733 goto out;
3734 }
3735 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3736 goto out;
3737 curagent->eps_max = atol(args[1]);
3738 }
3739 else if (!strcmp(args[0], "max-frame-size")) {
3740 if (!*args[1]) {
3741 ha_alert("parsing [%s:%d] : '%s' expects an integer argument.\n",
3742 file, linenum, args[0]);
3743 err_code |= ERR_ALERT | ERR_FATAL;
3744 goto out;
3745 }
3746 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3747 goto out;
3748 curagent->max_frame_size = atol(args[1]);
3749 if (curagent->max_frame_size < MIN_FRAME_SIZE ||
3750 curagent->max_frame_size > MAX_FRAME_SIZE) {
3751 ha_alert("parsing [%s:%d] : '%s' expects a positive integer argument in the range [%d, %d].\n",
3752 file, linenum, args[0], MIN_FRAME_SIZE, MAX_FRAME_SIZE);
3753 err_code |= ERR_ALERT | ERR_FATAL;
3754 goto out;
3755 }
3756 }
3757 else if (!strcmp(args[0], "max-waiting-frames")) {
3758 if (!*args[1]) {
3759 ha_alert("parsing [%s:%d] : '%s' expects an integer argument.\n",
3760 file, linenum, args[0]);
3761 err_code |= ERR_ALERT | ERR_FATAL;
3762 goto out;
3763 }
3764 if (alertif_too_many_args(1, file, linenum, args, &err_code))
3765 goto out;
3766 curagent->max_fpa = atol(args[1]);
3767 if (curagent->max_fpa < 1) {
3768 ha_alert("parsing [%s:%d] : '%s' expects a positive integer argument.\n",
3769 file, linenum, args[0]);
3770 err_code |= ERR_ALERT | ERR_FATAL;
3771 goto out;
3772 }
3773 }
3774 else if (!strcmp(args[0], "register-var-names")) {
3775 int cur_arg;
3776
3777 if (!*args[1]) {
3778 ha_alert("parsing [%s:%d] : '%s' expects one or more variable names.\n",
3779 file, linenum, args[0]);
3780 err_code |= ERR_ALERT | ERR_FATAL;
3781 goto out;
3782 }
3783 cur_arg = 1;
3784 while (*args[cur_arg]) {
3785 struct spoe_var_placeholder *vph;
3786
3787 if ((vph = calloc(1, sizeof(*vph))) == NULL) {
3788 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3789 err_code |= ERR_ALERT | ERR_ABORT;
3790 goto out;
3791 }
3792 if ((vph->name = strdup(args[cur_arg])) == NULL) {
3793 free(vph);
3794 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3795 err_code |= ERR_ALERT | ERR_ABORT;
3796 goto out;
3797 }
3798 LIST_ADDQ(&curvars, &vph->list);
3799 cur_arg++;
3800 }
3801 }
3802 else if (!strcmp(args[0], "log")) {
3803 char *errmsg = NULL;
3804
3805 if (!parse_logsrv(args, &curlogsrvs, (kwm == 1), &errmsg)) {
3806 ha_alert("parsing [%s:%d] : %s : %s\n", file, linenum, args[0], errmsg);
3807 err_code |= ERR_ALERT | ERR_FATAL;
3808 goto out;
3809 }
3810 }
3811 else if (*args[0]) {
3812 ha_alert("parsing [%s:%d] : unknown keyword '%s' in spoe-agent section.\n",
3813 file, linenum, args[0]);
3814 err_code |= ERR_ALERT | ERR_FATAL;
3815 goto out;
3816 }
3817 out:
3818 return err_code;
3819 }
3820 static int
cfg_parse_spoe_group(const char * file,int linenum,char ** args,int kwm)3821 cfg_parse_spoe_group(const char *file, int linenum, char **args, int kwm)
3822 {
3823 struct spoe_group *grp;
3824 const char *err;
3825 int err_code = 0;
3826
3827 if ((cfg_scope == NULL && curengine != NULL) ||
3828 (cfg_scope != NULL && curengine == NULL) ||
3829 (curengine != NULL && cfg_scope != NULL && strcmp(curengine, cfg_scope)))
3830 goto out;
3831
3832 if (!strcmp(args[0], "spoe-group")) { /* new spoe-group section */
3833 if (!*args[1]) {
3834 ha_alert("parsing [%s:%d] : missing name for spoe-group section.\n",
3835 file, linenum);
3836 err_code |= ERR_ALERT | ERR_ABORT;
3837 goto out;
3838 }
3839 if (alertif_too_many_args(1, file, linenum, args, &err_code)) {
3840 err_code |= ERR_ABORT;
3841 goto out;
3842 }
3843
3844 err = invalid_char(args[1]);
3845 if (err) {
3846 ha_alert("parsing [%s:%d] : character '%c' is not permitted in '%s' name '%s'.\n",
3847 file, linenum, *err, args[0], args[1]);
3848 err_code |= ERR_ALERT | ERR_ABORT;
3849 goto out;
3850 }
3851
3852 list_for_each_entry(grp, &curgrps, list) {
3853 if (!strcmp(grp->id, args[1])) {
3854 ha_alert("parsing [%s:%d]: spoe-group section '%s' has the same"
3855 " name as another one declared at %s:%d.\n",
3856 file, linenum, args[1], grp->conf.file, grp->conf.line);
3857 err_code |= ERR_ALERT | ERR_FATAL;
3858 goto out;
3859 }
3860 }
3861
3862 if ((curgrp = calloc(1, sizeof(*curgrp))) == NULL) {
3863 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3864 err_code |= ERR_ALERT | ERR_ABORT;
3865 goto out;
3866 }
3867
3868 curgrp->id = strdup(args[1]);
3869 curgrp->conf.file = strdup(file);
3870 curgrp->conf.line = linenum;
3871 LIST_INIT(&curgrp->phs);
3872 LIST_INIT(&curgrp->messages);
3873 LIST_ADDQ(&curgrps, &curgrp->list);
3874 }
3875 else if (!strcmp(args[0], "messages")) {
3876 int cur_arg = 1;
3877 while (*args[cur_arg]) {
3878 struct spoe_placeholder *ph = NULL;
3879
3880 list_for_each_entry(ph, &curgrp->phs, list) {
3881 if (!strcmp(ph->id, args[cur_arg])) {
3882 ha_alert("parsing [%s:%d]: spoe-message '%s' already used.\n",
3883 file, linenum, args[cur_arg]);
3884 err_code |= ERR_ALERT | ERR_FATAL;
3885 goto out;
3886 }
3887 }
3888
3889 if ((ph = calloc(1, sizeof(*ph))) == NULL) {
3890 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3891 err_code |= ERR_ALERT | ERR_ABORT;
3892 goto out;
3893 }
3894 ph->id = strdup(args[cur_arg]);
3895 LIST_ADDQ(&curgrp->phs, &ph->list);
3896 cur_arg++;
3897 }
3898 }
3899 else if (*args[0]) {
3900 ha_alert("parsing [%s:%d] : unknown keyword '%s' in spoe-group section.\n",
3901 file, linenum, args[0]);
3902 err_code |= ERR_ALERT | ERR_FATAL;
3903 goto out;
3904 }
3905 out:
3906 return err_code;
3907 }
3908
3909 static int
cfg_parse_spoe_message(const char * file,int linenum,char ** args,int kwm)3910 cfg_parse_spoe_message(const char *file, int linenum, char **args, int kwm)
3911 {
3912 struct spoe_message *msg;
3913 struct spoe_arg *arg;
3914 const char *err;
3915 char *errmsg = NULL;
3916 int err_code = 0;
3917
3918 if ((cfg_scope == NULL && curengine != NULL) ||
3919 (cfg_scope != NULL && curengine == NULL) ||
3920 (curengine != NULL && cfg_scope != NULL && strcmp(curengine, cfg_scope)))
3921 goto out;
3922
3923 if (!strcmp(args[0], "spoe-message")) { /* new spoe-message section */
3924 if (!*args[1]) {
3925 ha_alert("parsing [%s:%d] : missing name for spoe-message section.\n",
3926 file, linenum);
3927 err_code |= ERR_ALERT | ERR_ABORT;
3928 goto out;
3929 }
3930 if (alertif_too_many_args(1, file, linenum, args, &err_code)) {
3931 err_code |= ERR_ABORT;
3932 goto out;
3933 }
3934
3935 err = invalid_char(args[1]);
3936 if (err) {
3937 ha_alert("parsing [%s:%d] : character '%c' is not permitted in '%s' name '%s'.\n",
3938 file, linenum, *err, args[0], args[1]);
3939 err_code |= ERR_ALERT | ERR_ABORT;
3940 goto out;
3941 }
3942
3943 list_for_each_entry(msg, &curmsgs, list) {
3944 if (!strcmp(msg->id, args[1])) {
3945 ha_alert("parsing [%s:%d]: spoe-message section '%s' has the same"
3946 " name as another one declared at %s:%d.\n",
3947 file, linenum, args[1], msg->conf.file, msg->conf.line);
3948 err_code |= ERR_ALERT | ERR_FATAL;
3949 goto out;
3950 }
3951 }
3952
3953 if ((curmsg = calloc(1, sizeof(*curmsg))) == NULL) {
3954 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3955 err_code |= ERR_ALERT | ERR_ABORT;
3956 goto out;
3957 }
3958
3959 curmsg->id = strdup(args[1]);
3960 curmsg->id_len = strlen(curmsg->id);
3961 curmsg->event = SPOE_EV_NONE;
3962 curmsg->conf.file = strdup(file);
3963 curmsg->conf.line = linenum;
3964 curmsg->nargs = 0;
3965 LIST_INIT(&curmsg->args);
3966 LIST_INIT(&curmsg->acls);
3967 LIST_INIT(&curmsg->by_evt);
3968 LIST_INIT(&curmsg->by_grp);
3969 LIST_ADDQ(&curmsgs, &curmsg->list);
3970 }
3971 else if (!strcmp(args[0], "args")) {
3972 int cur_arg = 1;
3973
3974 curproxy->conf.args.ctx = ARGC_SPOE;
3975 curproxy->conf.args.file = file;
3976 curproxy->conf.args.line = linenum;
3977 while (*args[cur_arg]) {
3978 char *delim = strchr(args[cur_arg], '=');
3979 int idx = 0;
3980
3981 if ((arg = calloc(1, sizeof(*arg))) == NULL) {
3982 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3983 err_code |= ERR_ALERT | ERR_ABORT;
3984 goto out;
3985 }
3986
3987 if (!delim) {
3988 arg->name = NULL;
3989 arg->name_len = 0;
3990 delim = args[cur_arg];
3991 }
3992 else {
3993 arg->name = my_strndup(args[cur_arg], delim - args[cur_arg]);
3994 arg->name_len = delim - args[cur_arg];
3995 delim++;
3996 }
3997 arg->expr = sample_parse_expr((char*[]){delim, NULL},
3998 &idx, file, linenum, &errmsg,
3999 &curproxy->conf.args);
4000 if (arg->expr == NULL) {
4001 ha_alert("parsing [%s:%d] : '%s': %s.\n", file, linenum, args[0], errmsg);
4002 err_code |= ERR_ALERT | ERR_FATAL;
4003 free(arg->name);
4004 free(arg);
4005 goto out;
4006 }
4007 curmsg->nargs++;
4008 LIST_ADDQ(&curmsg->args, &arg->list);
4009 cur_arg++;
4010 }
4011 curproxy->conf.args.file = NULL;
4012 curproxy->conf.args.line = 0;
4013 }
4014 else if (!strcmp(args[0], "acl")) {
4015 err = invalid_char(args[1]);
4016 if (err) {
4017 ha_alert("parsing [%s:%d] : character '%c' is not permitted in acl name '%s'.\n",
4018 file, linenum, *err, args[1]);
4019 err_code |= ERR_ALERT | ERR_FATAL;
4020 goto out;
4021 }
4022 if (strcasecmp(args[1], "or") == 0) {
4023 ha_warning("parsing [%s:%d] : acl name '%s' will never match. 'or' is used to express a "
4024 "logical disjunction within a condition.\n",
4025 file, linenum, args[1]);
4026 err_code |= ERR_WARN;
4027 }
4028 if (parse_acl((const char **)args + 1, &curmsg->acls, &errmsg, &curproxy->conf.args, file, linenum) == NULL) {
4029 ha_alert("parsing [%s:%d] : error detected while parsing ACL '%s' : %s.\n",
4030 file, linenum, args[1], errmsg);
4031 err_code |= ERR_ALERT | ERR_FATAL;
4032 goto out;
4033 }
4034 }
4035 else if (!strcmp(args[0], "event")) {
4036 if (!*args[1]) {
4037 ha_alert("parsing [%s:%d] : missing event name.\n", file, linenum);
4038 err_code |= ERR_ALERT | ERR_FATAL;
4039 goto out;
4040 }
4041 /* if (alertif_too_many_args(1, file, linenum, args, &err_code)) */
4042 /* goto out; */
4043
4044 if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_CLIENT_SESS]))
4045 curmsg->event = SPOE_EV_ON_CLIENT_SESS;
4046 else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_SERVER_SESS]))
4047 curmsg->event = SPOE_EV_ON_SERVER_SESS;
4048
4049 else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_TCP_REQ_FE]))
4050 curmsg->event = SPOE_EV_ON_TCP_REQ_FE;
4051 else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_TCP_REQ_BE]))
4052 curmsg->event = SPOE_EV_ON_TCP_REQ_BE;
4053 else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_TCP_RSP]))
4054 curmsg->event = SPOE_EV_ON_TCP_RSP;
4055
4056 else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_HTTP_REQ_FE]))
4057 curmsg->event = SPOE_EV_ON_HTTP_REQ_FE;
4058 else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_HTTP_REQ_BE]))
4059 curmsg->event = SPOE_EV_ON_HTTP_REQ_BE;
4060 else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_HTTP_RSP]))
4061 curmsg->event = SPOE_EV_ON_HTTP_RSP;
4062 else {
4063 ha_alert("parsing [%s:%d] : unknown event '%s'.\n",
4064 file, linenum, args[1]);
4065 err_code |= ERR_ALERT | ERR_FATAL;
4066 goto out;
4067 }
4068
4069 if (strcmp(args[2], "if") == 0 || strcmp(args[2], "unless") == 0) {
4070 struct acl_cond *cond;
4071
4072 cond = build_acl_cond(file, linenum, &curmsg->acls,
4073 curproxy, (const char **)args+2,
4074 &errmsg);
4075 if (cond == NULL) {
4076 ha_alert("parsing [%s:%d] : error detected while "
4077 "parsing an 'event %s' condition : %s.\n",
4078 file, linenum, args[1], errmsg);
4079 err_code |= ERR_ALERT | ERR_FATAL;
4080 goto out;
4081 }
4082 curmsg->cond = cond;
4083 }
4084 else if (*args[2]) {
4085 ha_alert("parsing [%s:%d]: 'event %s' expects either 'if' "
4086 "or 'unless' followed by a condition but found '%s'.\n",
4087 file, linenum, args[1], args[2]);
4088 err_code |= ERR_ALERT | ERR_FATAL;
4089 goto out;
4090 }
4091 }
4092 else if (!*args[0]) {
4093 ha_alert("parsing [%s:%d] : unknown keyword '%s' in spoe-message section.\n",
4094 file, linenum, args[0]);
4095 err_code |= ERR_ALERT | ERR_FATAL;
4096 goto out;
4097 }
4098 out:
4099 free(errmsg);
4100 return err_code;
4101 }
4102
4103 /* Return -1 on error, else 0 */
4104 static int
parse_spoe_flt(char ** args,int * cur_arg,struct proxy * px,struct flt_conf * fconf,char ** err,void * private)4105 parse_spoe_flt(char **args, int *cur_arg, struct proxy *px,
4106 struct flt_conf *fconf, char **err, void *private)
4107 {
4108 struct list backup_sections;
4109 struct spoe_config *conf;
4110 struct spoe_message *msg, *msgback;
4111 struct spoe_group *grp, *grpback;
4112 struct spoe_placeholder *ph, *phback;
4113 struct spoe_var_placeholder *vph, *vphback;
4114 struct logsrv *logsrv, *logsrvback;
4115 char *file = NULL, *engine = NULL;
4116 int ret, pos = *cur_arg + 1;
4117
4118 LIST_INIT(&curmsgs);
4119 LIST_INIT(&curgrps);
4120 LIST_INIT(&curmphs);
4121 LIST_INIT(&curgphs);
4122 LIST_INIT(&curvars);
4123 LIST_INIT(&curlogsrvs);
4124 curpxopts = 0;
4125 curpxopts2 = 0;
4126
4127 conf = calloc(1, sizeof(*conf));
4128 if (conf == NULL) {
4129 memprintf(err, "%s: out of memory", args[*cur_arg]);
4130 goto error;
4131 }
4132 conf->proxy = px;
4133
4134 while (*args[pos]) {
4135 if (!strcmp(args[pos], "config")) {
4136 if (!*args[pos+1]) {
4137 memprintf(err, "'%s' : '%s' option without value",
4138 args[*cur_arg], args[pos]);
4139 goto error;
4140 }
4141 file = args[pos+1];
4142 pos += 2;
4143 }
4144 else if (!strcmp(args[pos], "engine")) {
4145 if (!*args[pos+1]) {
4146 memprintf(err, "'%s' : '%s' option without value",
4147 args[*cur_arg], args[pos]);
4148 goto error;
4149 }
4150 engine = args[pos+1];
4151 pos += 2;
4152 }
4153 else {
4154 memprintf(err, "unknown keyword '%s'", args[pos]);
4155 goto error;
4156 }
4157 }
4158 if (file == NULL) {
4159 memprintf(err, "'%s' : missing config file", args[*cur_arg]);
4160 goto error;
4161 }
4162
4163 /* backup sections and register SPOE sections */
4164 LIST_INIT(&backup_sections);
4165 cfg_backup_sections(&backup_sections);
4166 cfg_register_section("spoe-agent", cfg_parse_spoe_agent, NULL);
4167 cfg_register_section("spoe-group", cfg_parse_spoe_group, NULL);
4168 cfg_register_section("spoe-message", cfg_parse_spoe_message, NULL);
4169
4170 /* Parse SPOE filter configuration file */
4171 curengine = engine;
4172 curproxy = px;
4173 curagent = NULL;
4174 curmsg = NULL;
4175 ret = readcfgfile(file);
4176 curproxy = NULL;
4177
4178 /* unregister SPOE sections and restore previous sections */
4179 cfg_unregister_sections();
4180 cfg_restore_sections(&backup_sections);
4181
4182 if (ret == -1) {
4183 memprintf(err, "Could not open configuration file %s : %s",
4184 file, strerror(errno));
4185 goto error;
4186 }
4187 if (ret & (ERR_ABORT|ERR_FATAL)) {
4188 memprintf(err, "Error(s) found in configuration file %s", file);
4189 goto error;
4190 }
4191
4192 /* Check SPOE agent */
4193 if (curagent == NULL) {
4194 memprintf(err, "No SPOE agent found in file %s", file);
4195 goto error;
4196 }
4197 if (curagent->b.name == NULL) {
4198 memprintf(err, "No backend declared for SPOE agent '%s' declared at %s:%d",
4199 curagent->id, curagent->conf.file, curagent->conf.line);
4200 goto error;
4201 }
4202 if (curagent->timeout.hello == TICK_ETERNITY ||
4203 curagent->timeout.idle == TICK_ETERNITY ||
4204 curagent->timeout.processing == TICK_ETERNITY) {
4205 ha_warning("Proxy '%s': missing timeouts for SPOE agent '%s' declare at %s:%d.\n"
4206 " | While not properly invalid, you will certainly encounter various problems\n"
4207 " | with such a configuration. To fix this, please ensure that all following\n"
4208 " | timeouts are set to a non-zero value: 'hello', 'idle', 'processing'.\n",
4209 px->id, curagent->id, curagent->conf.file, curagent->conf.line);
4210 }
4211 if (curagent->var_pfx == NULL) {
4212 char *tmp = curagent->id;
4213
4214 while (*tmp) {
4215 if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
4216 memprintf(err, "Invalid variable prefix '%s' for SPOE agent '%s' declared at %s:%d. "
4217 "Use 'option var-prefix' to set it. Only [a-zA-Z0-9_.] chars are supported.\n",
4218 curagent->id, curagent->id, curagent->conf.file, curagent->conf.line);
4219 goto error;
4220 }
4221 tmp++;
4222 }
4223 curagent->var_pfx = strdup(curagent->id);
4224 }
4225
4226 if (curagent->var_on_error) {
4227 struct arg arg;
4228
4229 trash.data = snprintf(trash.area, trash.size, "txn.%s.%s",
4230 curagent->var_pfx, curagent->var_on_error);
4231
4232 arg.type = ARGT_STR;
4233 arg.data.str.area = trash.area;
4234 arg.data.str.data = trash.data;
4235 arg.data.str.size = 0; /* Set it to 0 to not release it in vars_check_args() */
4236 if (!vars_check_arg(&arg, err)) {
4237 memprintf(err, "SPOE agent '%s': failed to register variable %s.%s (%s)",
4238 curagent->id, curagent->var_pfx, curagent->var_on_error, *err);
4239 goto error;
4240 }
4241 }
4242
4243 if (curagent->var_t_process) {
4244 struct arg arg;
4245
4246 trash.data = snprintf(trash.area, trash.size, "txn.%s.%s",
4247 curagent->var_pfx, curagent->var_t_process);
4248
4249 arg.type = ARGT_STR;
4250 arg.data.str.area = trash.area;
4251 arg.data.str.data = trash.data;
4252 arg.data.str.size = 0; /* Set it to 0 to not release it in vars_check_args() */
4253 if (!vars_check_arg(&arg, err)) {
4254 memprintf(err, "SPOE agent '%s': failed to register variable %s.%s (%s)",
4255 curagent->id, curagent->var_pfx, curagent->var_t_process, *err);
4256 goto error;
4257 }
4258 }
4259
4260 if (curagent->var_t_total) {
4261 struct arg arg;
4262
4263 trash.data = snprintf(trash.area, trash.size, "txn.%s.%s",
4264 curagent->var_pfx, curagent->var_t_total);
4265
4266 arg.type = ARGT_STR;
4267 arg.data.str.area = trash.area;
4268 arg.data.str.data = trash.data;
4269 arg.data.str.size = 0; /* Set it to 0 to not release it in vars_check_args() */
4270 if (!vars_check_arg(&arg, err)) {
4271 memprintf(err, "SPOE agent '%s': failed to register variable %s.%s (%s)",
4272 curagent->id, curagent->var_pfx, curagent->var_t_process, *err);
4273 goto error;
4274 }
4275 }
4276
4277 if (LIST_ISEMPTY(&curmphs) && LIST_ISEMPTY(&curgphs)) {
4278 ha_warning("Proxy '%s': No message/group used by SPOE agent '%s' declared at %s:%d.\n",
4279 px->id, curagent->id, curagent->conf.file, curagent->conf.line);
4280 goto finish;
4281 }
4282
4283 /* Replace placeholders by the corresponding messages for the SPOE
4284 * agent */
4285 list_for_each_entry(ph, &curmphs, list) {
4286 list_for_each_entry(msg, &curmsgs, list) {
4287 struct spoe_arg *arg;
4288 unsigned int where;
4289
4290 if (!strcmp(msg->id, ph->id)) {
4291 if ((px->cap & (PR_CAP_FE|PR_CAP_BE)) == (PR_CAP_FE|PR_CAP_BE)) {
4292 if (msg->event == SPOE_EV_ON_TCP_REQ_BE)
4293 msg->event = SPOE_EV_ON_TCP_REQ_FE;
4294 if (msg->event == SPOE_EV_ON_HTTP_REQ_BE)
4295 msg->event = SPOE_EV_ON_HTTP_REQ_FE;
4296 }
4297 if (!(px->cap & PR_CAP_FE) && (msg->event == SPOE_EV_ON_CLIENT_SESS ||
4298 msg->event == SPOE_EV_ON_TCP_REQ_FE ||
4299 msg->event == SPOE_EV_ON_HTTP_REQ_FE)) {
4300 ha_warning("Proxy '%s': frontend event used on a backend proxy at %s:%d.\n",
4301 px->id, msg->conf.file, msg->conf.line);
4302 goto next_mph;
4303 }
4304 if (msg->event == SPOE_EV_NONE) {
4305 ha_warning("Proxy '%s': Ignore SPOE message '%s' without event at %s:%d.\n",
4306 px->id, msg->id, msg->conf.file, msg->conf.line);
4307 goto next_mph;
4308 }
4309
4310 where = 0;
4311 switch (msg->event) {
4312 case SPOE_EV_ON_CLIENT_SESS:
4313 where |= SMP_VAL_FE_CON_ACC;
4314 break;
4315
4316 case SPOE_EV_ON_TCP_REQ_FE:
4317 where |= SMP_VAL_FE_REQ_CNT;
4318 break;
4319
4320 case SPOE_EV_ON_HTTP_REQ_FE:
4321 where |= SMP_VAL_FE_HRQ_HDR;
4322 break;
4323
4324 case SPOE_EV_ON_TCP_REQ_BE:
4325 if (px->cap & PR_CAP_FE)
4326 where |= SMP_VAL_FE_REQ_CNT;
4327 if (px->cap & PR_CAP_BE)
4328 where |= SMP_VAL_BE_REQ_CNT;
4329 break;
4330
4331 case SPOE_EV_ON_HTTP_REQ_BE:
4332 if (px->cap & PR_CAP_FE)
4333 where |= SMP_VAL_FE_HRQ_HDR;
4334 if (px->cap & PR_CAP_BE)
4335 where |= SMP_VAL_BE_HRQ_HDR;
4336 break;
4337
4338 case SPOE_EV_ON_SERVER_SESS:
4339 where |= SMP_VAL_BE_SRV_CON;
4340 break;
4341
4342 case SPOE_EV_ON_TCP_RSP:
4343 if (px->cap & PR_CAP_FE)
4344 where |= SMP_VAL_FE_RES_CNT;
4345 if (px->cap & PR_CAP_BE)
4346 where |= SMP_VAL_BE_RES_CNT;
4347 break;
4348
4349 case SPOE_EV_ON_HTTP_RSP:
4350 if (px->cap & PR_CAP_FE)
4351 where |= SMP_VAL_FE_HRS_HDR;
4352 if (px->cap & PR_CAP_BE)
4353 where |= SMP_VAL_BE_HRS_HDR;
4354 break;
4355
4356 default:
4357 break;
4358 }
4359
4360 list_for_each_entry(arg, &msg->args, list) {
4361 if (!(arg->expr->fetch->val & where)) {
4362 memprintf(err, "Ignore SPOE message '%s' at %s:%d: "
4363 "some args extract information from '%s', "
4364 "none of which is available here ('%s')",
4365 msg->id, msg->conf.file, msg->conf.line,
4366 sample_ckp_names(arg->expr->fetch->use),
4367 sample_ckp_names(where));
4368 goto error;
4369 }
4370 }
4371
4372 msg->agent = curagent;
4373 LIST_ADDQ(&curagent->events[msg->event], &msg->by_evt);
4374 goto next_mph;
4375 }
4376 }
4377 memprintf(err, "SPOE agent '%s' try to use undefined SPOE message '%s' at %s:%d",
4378 curagent->id, ph->id, curagent->conf.file, curagent->conf.line);
4379 goto error;
4380 next_mph:
4381 continue;
4382 }
4383
4384 /* Replace placeholders by the corresponding groups for the SPOE
4385 * agent */
4386 list_for_each_entry(ph, &curgphs, list) {
4387 list_for_each_entry_safe(grp, grpback, &curgrps, list) {
4388 if (!strcmp(grp->id, ph->id)) {
4389 grp->agent = curagent;
4390 LIST_DEL(&grp->list);
4391 LIST_ADDQ(&curagent->groups, &grp->list);
4392 goto next_aph;
4393 }
4394 }
4395 memprintf(err, "SPOE agent '%s' try to use undefined SPOE group '%s' at %s:%d",
4396 curagent->id, ph->id, curagent->conf.file, curagent->conf.line);
4397 goto error;
4398 next_aph:
4399 continue;
4400 }
4401
4402 /* Replace placeholders by the corresponding message for each SPOE
4403 * group of the SPOE agent */
4404 list_for_each_entry(grp, &curagent->groups, list) {
4405 list_for_each_entry_safe(ph, phback, &grp->phs, list) {
4406 list_for_each_entry(msg, &curmsgs, list) {
4407 if (!strcmp(msg->id, ph->id)) {
4408 if (msg->group != NULL) {
4409 memprintf(err, "SPOE message '%s' already belongs to "
4410 "the SPOE group '%s' declare at %s:%d",
4411 msg->id, msg->group->id,
4412 msg->group->conf.file,
4413 msg->group->conf.line);
4414 goto error;
4415 }
4416
4417 /* Scope for arguments are not checked for now. We will check
4418 * them only if a rule use the corresponding SPOE group. */
4419 msg->agent = curagent;
4420 msg->group = grp;
4421 LIST_DEL(&ph->list);
4422 LIST_ADDQ(&grp->messages, &msg->by_grp);
4423 goto next_mph_grp;
4424 }
4425 }
4426 memprintf(err, "SPOE group '%s' try to use undefined SPOE message '%s' at %s:%d",
4427 grp->id, ph->id, curagent->conf.file, curagent->conf.line);
4428 goto error;
4429 next_mph_grp:
4430 continue;
4431 }
4432 }
4433
4434 finish:
4435 /* move curmsgs to the agent message list */
4436 curmsgs.n->p = &curagent->messages;
4437 curmsgs.p->n = &curagent->messages;
4438 curagent->messages = curmsgs;
4439 LIST_INIT(&curmsgs);
4440
4441 conf->id = strdup(engine ? engine : curagent->id);
4442 conf->agent = curagent;
4443
4444 /* Start agent's proxy initialization here. It will be finished during
4445 * the filter init. */
4446 memset(&conf->agent_fe, 0, sizeof(conf->agent_fe));
4447 init_new_proxy(&conf->agent_fe);
4448 conf->agent_fe.id = conf->agent->id;
4449 conf->agent_fe.parent = conf->agent;
4450 conf->agent_fe.options |= curpxopts;
4451 conf->agent_fe.options2 |= curpxopts2;
4452
4453 list_for_each_entry_safe(logsrv, logsrvback, &curlogsrvs, list) {
4454 LIST_DEL(&logsrv->list);
4455 LIST_ADDQ(&conf->agent_fe.logsrvs, &logsrv->list);
4456 }
4457
4458 list_for_each_entry_safe(ph, phback, &curmphs, list) {
4459 LIST_DEL(&ph->list);
4460 spoe_release_placeholder(ph);
4461 }
4462 list_for_each_entry_safe(ph, phback, &curgphs, list) {
4463 LIST_DEL(&ph->list);
4464 spoe_release_placeholder(ph);
4465 }
4466 list_for_each_entry_safe(vph, vphback, &curvars, list) {
4467 struct arg arg;
4468
4469 trash.data = snprintf(trash.area, trash.size, "proc.%s.%s",
4470 curagent->var_pfx, vph->name);
4471
4472 arg.type = ARGT_STR;
4473 arg.data.str.area = trash.area;
4474 arg.data.str.data = trash.data;
4475 arg.data.str.size = 0; /* Set it to 0 to not release it in vars_check_args() */
4476 if (!vars_check_arg(&arg, err)) {
4477 memprintf(err, "SPOE agent '%s': failed to register variable %s.%s (%s)",
4478 curagent->id, curagent->var_pfx, vph->name, *err);
4479 goto error;
4480 }
4481
4482 LIST_DEL(&vph->list);
4483 free(vph->name);
4484 free(vph);
4485 }
4486 list_for_each_entry_safe(grp, grpback, &curgrps, list) {
4487 LIST_DEL(&grp->list);
4488 spoe_release_group(grp);
4489 }
4490 *cur_arg = pos;
4491 fconf->id = spoe_filter_id;
4492 fconf->ops = &spoe_ops;
4493 fconf->conf = conf;
4494 return 0;
4495
4496 error:
4497 spoe_release_agent(curagent);
4498 list_for_each_entry_safe(ph, phback, &curmphs, list) {
4499 LIST_DEL(&ph->list);
4500 spoe_release_placeholder(ph);
4501 }
4502 list_for_each_entry_safe(ph, phback, &curgphs, list) {
4503 LIST_DEL(&ph->list);
4504 spoe_release_placeholder(ph);
4505 }
4506 list_for_each_entry_safe(vph, vphback, &curvars, list) {
4507 LIST_DEL(&vph->list);
4508 free(vph->name);
4509 free(vph);
4510 }
4511 list_for_each_entry_safe(grp, grpback, &curgrps, list) {
4512 LIST_DEL(&grp->list);
4513 spoe_release_group(grp);
4514 }
4515 list_for_each_entry_safe(msg, msgback, &curmsgs, list) {
4516 LIST_DEL(&msg->list);
4517 spoe_release_message(msg);
4518 }
4519 list_for_each_entry_safe(logsrv, logsrvback, &curlogsrvs, list) {
4520 LIST_DEL(&logsrv->list);
4521 free(logsrv);
4522 }
4523 free(conf);
4524 return -1;
4525 }
4526
4527 /* Send message of a SPOE group. This is the action_ptr callback of a rule
4528 * associated to a "send-spoe-group" action.
4529 *
4530 * It returns ACT_RET_CONT is processing is finished without error, it returns
4531 * ACT_RET_YIELD if the action is in progress. Otherwise it returns
4532 * ACT_RET_ERR. */
4533 static enum act_return
spoe_send_group(struct act_rule * rule,struct proxy * px,struct session * sess,struct stream * s,int flags)4534 spoe_send_group(struct act_rule *rule, struct proxy *px,
4535 struct session *sess, struct stream *s, int flags)
4536 {
4537 struct filter *filter;
4538 struct spoe_agent *agent = NULL;
4539 struct spoe_group *group = NULL;
4540 struct spoe_context *ctx = NULL;
4541 int ret, dir;
4542
4543 list_for_each_entry(filter, &s->strm_flt.filters, list) {
4544 if (filter->config == rule->arg.act.p[0]) {
4545 agent = rule->arg.act.p[2];
4546 group = rule->arg.act.p[3];
4547 ctx = filter->ctx;
4548 break;
4549 }
4550 }
4551 if (agent == NULL || group == NULL || ctx == NULL)
4552 return ACT_RET_ERR;
4553 if (ctx->state == SPOE_CTX_ST_NONE)
4554 return ACT_RET_CONT;
4555
4556 switch (rule->from) {
4557 case ACT_F_TCP_REQ_SES: dir = SMP_OPT_DIR_REQ; break;
4558 case ACT_F_TCP_REQ_CNT: dir = SMP_OPT_DIR_REQ; break;
4559 case ACT_F_TCP_RES_CNT: dir = SMP_OPT_DIR_RES; break;
4560 case ACT_F_HTTP_REQ: dir = SMP_OPT_DIR_REQ; break;
4561 case ACT_F_HTTP_RES: dir = SMP_OPT_DIR_RES; break;
4562 default:
4563 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
4564 " - internal error while execute spoe-send-group\n",
4565 (int)now.tv_sec, (int)now.tv_usec, agent->id,
4566 __FUNCTION__, s);
4567 send_log(px, LOG_ERR, "SPOE: [%s] internal error while execute spoe-send-group\n",
4568 agent->id);
4569 return ACT_RET_CONT;
4570 }
4571
4572 ret = spoe_process_group(s, ctx, group, dir);
4573 if (ret == 1)
4574 return ACT_RET_CONT;
4575 else if (ret == 0) {
4576 if (flags & ACT_FLAG_FINAL) {
4577 SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
4578 " - failed to process group '%s': interrupted by caller\n",
4579 (int)now.tv_sec, (int)now.tv_usec,
4580 agent->id, __FUNCTION__, s, group->id);
4581 ctx->status_code = SPOE_CTX_ERR_INTERRUPT;
4582 spoe_stop_processing(agent, ctx);
4583 spoe_handle_processing_error(s, agent, ctx, dir);
4584 return ACT_RET_CONT;
4585 }
4586 return ACT_RET_YIELD;
4587 }
4588 else
4589 return ACT_RET_ERR;
4590 }
4591
4592 /* Check an "send-spoe-group" action. Here, we'll try to find the real SPOE
4593 * group associated to <rule>. The format of an rule using 'send-spoe-group'
4594 * action should be:
4595 *
4596 * (http|tcp)-(request|response) send-spoe-group <engine-id> <group-id>
4597 *
4598 * So, we'll loop on each configured SPOE filter for the proxy <px> to find the
4599 * SPOE engine matching <engine-id>. And then, we'll try to find the good group
4600 * matching <group-id>. Finally, we'll check all messages referenced by the SPOE
4601 * group.
4602 *
4603 * The function returns 1 in success case, otherwise, it returns 0 and err is
4604 * filled.
4605 */
4606 static int
check_send_spoe_group(struct act_rule * rule,struct proxy * px,char ** err)4607 check_send_spoe_group(struct act_rule *rule, struct proxy *px, char **err)
4608 {
4609 struct flt_conf *fconf;
4610 struct spoe_config *conf;
4611 struct spoe_agent *agent = NULL;
4612 struct spoe_group *group;
4613 struct spoe_message *msg;
4614 char *engine_id = rule->arg.act.p[0];
4615 char *group_id = rule->arg.act.p[1];
4616 unsigned int where = 0;
4617
4618 switch (rule->from) {
4619 case ACT_F_TCP_REQ_SES: where = SMP_VAL_FE_SES_ACC; break;
4620 case ACT_F_TCP_REQ_CNT: where = SMP_VAL_FE_REQ_CNT; break;
4621 case ACT_F_TCP_RES_CNT: where = SMP_VAL_BE_RES_CNT; break;
4622 case ACT_F_HTTP_REQ: where = SMP_VAL_FE_HRQ_HDR; break;
4623 case ACT_F_HTTP_RES: where = SMP_VAL_BE_HRS_HDR; break;
4624 default:
4625 memprintf(err,
4626 "internal error, unexpected rule->from=%d, please report this bug!",
4627 rule->from);
4628 goto error;
4629 }
4630
4631 /* Try to find the SPOE engine by checking all SPOE filters for proxy
4632 * <px> */
4633 list_for_each_entry(fconf, &px->filter_configs, list) {
4634 conf = fconf->conf;
4635
4636 /* This is not an SPOE filter */
4637 if (fconf->id != spoe_filter_id)
4638 continue;
4639
4640 /* This is the good engine */
4641 if (!strcmp(conf->id, engine_id)) {
4642 agent = conf->agent;
4643 break;
4644 }
4645 }
4646 if (agent == NULL) {
4647 memprintf(err, "unable to find SPOE engine '%s' used by the send-spoe-group '%s'",
4648 engine_id, group_id);
4649 goto error;
4650 }
4651
4652 /* Try to find the right group */
4653 list_for_each_entry(group, &agent->groups, list) {
4654 /* This is the good group */
4655 if (!strcmp(group->id, group_id))
4656 break;
4657 }
4658 if (&group->list == &agent->groups) {
4659 memprintf(err, "unable to find SPOE group '%s' into SPOE engine '%s' configuration",
4660 group_id, engine_id);
4661 goto error;
4662 }
4663
4664 /* Ok, we found the group, we need to check messages and their
4665 * arguments */
4666 list_for_each_entry(msg, &group->messages, by_grp) {
4667 struct spoe_arg *arg;
4668
4669 list_for_each_entry(arg, &msg->args, list) {
4670 if (!(arg->expr->fetch->val & where)) {
4671 memprintf(err, "Invalid SPOE message '%s' used by SPOE group '%s' at %s:%d: "
4672 "some args extract information from '%s',"
4673 "none of which is available here ('%s')",
4674 msg->id, group->id, msg->conf.file, msg->conf.line,
4675 sample_ckp_names(arg->expr->fetch->use),
4676 sample_ckp_names(where));
4677 goto error;
4678 }
4679 }
4680 }
4681
4682 free(engine_id);
4683 free(group_id);
4684 rule->arg.act.p[0] = fconf; /* Associate filter config with the rule */
4685 rule->arg.act.p[1] = conf; /* Associate SPOE config with the rule */
4686 rule->arg.act.p[2] = agent; /* Associate SPOE agent with the rule */
4687 rule->arg.act.p[3] = group; /* Associate SPOE group with the rule */
4688 return 1;
4689
4690 error:
4691 free(engine_id);
4692 free(group_id);
4693 return 0;
4694 }
4695
4696 /* Parse 'send-spoe-group' action following the format:
4697 *
4698 * ... send-spoe-group <engine-id> <group-id>
4699 *
4700 * It returns ACT_RET_PRS_ERR if fails and <err> is filled with an error
4701 * message. Otherwise, it returns ACT_RET_PRS_OK and parsing engine and group
4702 * ids are saved and used later, when the rule will be checked.
4703 */
4704 static enum act_parse_ret
parse_send_spoe_group(const char ** args,int * orig_arg,struct proxy * px,struct act_rule * rule,char ** err)4705 parse_send_spoe_group(const char **args, int *orig_arg, struct proxy *px,
4706 struct act_rule *rule, char **err)
4707 {
4708 if (!*args[*orig_arg] || !*args[*orig_arg+1] ||
4709 (*args[*orig_arg+2] && strcmp(args[*orig_arg+2], "if") != 0 && strcmp(args[*orig_arg+2], "unless") != 0)) {
4710 memprintf(err, "expects 2 arguments: <engine-id> <group-id>");
4711 return ACT_RET_PRS_ERR;
4712 }
4713 rule->arg.act.p[0] = strdup(args[*orig_arg]); /* Copy the SPOE engine id */
4714 rule->arg.act.p[1] = strdup(args[*orig_arg+1]); /* Cope the SPOE group id */
4715
4716 (*orig_arg) += 2;
4717
4718 rule->action = ACT_CUSTOM;
4719 rule->action_ptr = spoe_send_group;
4720 rule->check_ptr = check_send_spoe_group;
4721 return ACT_RET_PRS_OK;
4722 }
4723
4724
4725 /* Declare the filter parser for "spoe" keyword */
4726 static struct flt_kw_list flt_kws = { "SPOE", { }, {
4727 { "spoe", parse_spoe_flt, NULL },
4728 { NULL, NULL, NULL },
4729 }
4730 };
4731
4732 INITCALL1(STG_REGISTER, flt_register_keywords, &flt_kws);
4733
4734 /* Delcate the action parser for "spoe-action" keyword */
4735 static struct action_kw_list tcp_req_action_kws = { { }, {
4736 { "send-spoe-group", parse_send_spoe_group },
4737 { /* END */ },
4738 }
4739 };
4740
4741 INITCALL1(STG_REGISTER, tcp_req_cont_keywords_register, &tcp_req_action_kws);
4742
4743 static struct action_kw_list tcp_res_action_kws = { { }, {
4744 { "send-spoe-group", parse_send_spoe_group },
4745 { /* END */ },
4746 }
4747 };
4748
4749 INITCALL1(STG_REGISTER, tcp_res_cont_keywords_register, &tcp_res_action_kws);
4750
4751 static struct action_kw_list http_req_action_kws = { { }, {
4752 { "send-spoe-group", parse_send_spoe_group },
4753 { /* END */ },
4754 }
4755 };
4756
4757 INITCALL1(STG_REGISTER, http_req_keywords_register, &http_req_action_kws);
4758
4759 static struct action_kw_list http_res_action_kws = { { }, {
4760 { "send-spoe-group", parse_send_spoe_group },
4761 { /* END */ },
4762 }
4763 };
4764
4765 INITCALL1(STG_REGISTER, http_res_keywords_register, &http_res_action_kws);
4766