RepoMirrors
/
haproxy
mirror of http://git.haproxy.org/git/haproxy.git/
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
haproxy/contrib/spoa_server/spoa.c

1317 lines
31 KiB
C
Raw Blame History

/*
* A Random IP reputation service acting as a Stream Processing Offload Agent
*
* This is a very simple service that implement a "random" ip reputation
* service. It will return random scores for all checked IP addresses. It only
* shows you how to implement a ip reputation service or such kind of services
* using the SPOE.
*
* Copyright 2016 HAProxy Technologies, Christopher Faulet <cfaulet@haproxy.com>
* Copyright 2018 OZON / Thierry Fournier <thierry.fournier@ozon.io>
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version
* 2 of the License, or (at your option) any later version.
*
*/
#include <limits.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>
#include <unistd.h>
#include <signal.h>
#include <pthread.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include "spoa.h"
#define DEFAULT_PORT 12345
#define NUM_WORKERS 5
#define SLEN(str) (sizeof(str)-1)
/* Frame Types sent by HAProxy and by agents */
enum spoe_frame_type {
/* Frames sent by HAProxy */
SPOE_FRM_T_HAPROXY_HELLO = 1,
SPOE_FRM_T_HAPROXY_DISCON,
SPOE_FRM_T_HAPROXY_NOTIFY,
/* Frames sent by the agents */
SPOE_FRM_T_AGENT_HELLO = 101,
SPOE_FRM_T_AGENT_DISCON,
SPOE_FRM_T_AGENT_ACK
};
/* Errors triggered by SPOE applet */
enum spoe_frame_error {
SPOE_FRM_ERR_NONE = 0,
SPOE_FRM_ERR_IO,
SPOE_FRM_ERR_TOUT,
SPOE_FRM_ERR_TOO_BIG,
SPOE_FRM_ERR_INVALID,
SPOE_FRM_ERR_NO_VSN,
SPOE_FRM_ERR_NO_FRAME_SIZE,
SPOE_FRM_ERR_NO_CAP,
SPOE_FRM_ERR_BAD_VSN,
SPOE_FRM_ERR_BAD_FRAME_SIZE,
SPOE_FRM_ERR_UNKNOWN = 99,
SPOE_FRM_ERRS,
};
/* All supported SPOE actions */
enum spoe_action_type {
SPOE_ACT_T_SET_VAR = 1,
SPOE_ACT_T_UNSET_VAR,
SPOE_ACT_TYPES,
};
/* Masks to get data type or flags value */
#define SPOE_DATA_T_MASK 0x0F
#define SPOE_DATA_FL_MASK 0xF0
/* Flags to set Boolean values */
#define SPOE_DATA_FL_FALSE 0x00
#define SPOE_DATA_FL_TRUE 0x10
static const char *spoe_frm_err_reasons[SPOE_FRM_ERRS] = {
[SPOE_FRM_ERR_NONE] = "normal",
[SPOE_FRM_ERR_IO] = "I/O error",
[SPOE_FRM_ERR_TOUT] = "a timeout occurred",
[SPOE_FRM_ERR_TOO_BIG] = "frame is too big",
[SPOE_FRM_ERR_INVALID] = "invalid frame received",
[SPOE_FRM_ERR_NO_VSN] = "version value not found",
[SPOE_FRM_ERR_NO_FRAME_SIZE] = "max-frame-size value not found",
[SPOE_FRM_ERR_NO_CAP] = "capabilities value not found",
[SPOE_FRM_ERR_BAD_VSN] = "unsupported version",
[SPOE_FRM_ERR_BAD_FRAME_SIZE] = "max-frame-size too big or too small",
[SPOE_FRM_ERR_UNKNOWN] = "an unknown error occurred",
};
bool debug = false;
pthread_key_t worker_id;
static struct ps *ps_list = NULL;
static struct ps_message *ps_messages = NULL;
static int nfiles = 0;
static char **files = NULL;
static inline void add_file(const char *file)
{
nfiles++;
files = realloc(files, sizeof(*files) * nfiles);
if (files == NULL) {
fprintf(stderr, "Out of memory error\n");
exit(EXIT_FAILURE);
}
files[nfiles - 1] = strdup(file);
if (files[nfiles - 1] == NULL) {
fprintf(stderr, "Out of memory error\n");
exit(EXIT_FAILURE);
}
}
void ps_register(struct ps *ps)
{
ps->next = ps_list;
ps_list = ps;
}
void ps_register_message(struct ps *ps, const char *name, void *ref)
{
struct ps_message *msg;
/* Look for already registered name */
for (msg = ps_messages; msg; msg = msg->next) {
if (strcmp(name, msg->name) == 0) {
LOG("Message \"%s\" already registered\n", name);
exit(EXIT_FAILURE);
}
}
msg = calloc(1, sizeof(*msg));
if (msg == NULL) {
LOG("Out of memory error\n");
exit(EXIT_FAILURE);
}
msg->next = ps_messages;
ps_messages = msg;
msg->name = strdup(name);
if (msg->name == NULL) {
LOG("Out of memory error\n");
exit(EXIT_FAILURE);
}
msg->ref = ref;
msg->ps = ps;
}
static int
do_read(int sock, void *buf, int read_len)
{
fd_set readfds;
int n = 0, total = 0, bytesleft = read_len;
FD_ZERO(&readfds);
FD_SET(sock, &readfds);
while (total < read_len) {
if (select(FD_SETSIZE, &readfds, NULL, NULL, NULL) == -1)
return -1;
if (!FD_ISSET(sock, &readfds))
return -1;
n = read(sock, buf + total, bytesleft);
if (n <= 0)
break;
total += n;
bytesleft -= n;
}
return (n == -1) ? -1 : total;
}
static int
do_write(int sock, void *buf, int write_len)
{
fd_set writefds;
int n = 0, total = 0, bytesleft = write_len;
FD_ZERO(&writefds);
FD_SET(sock, &writefds);
while (total < write_len) {
if (select(FD_SETSIZE, NULL, &writefds, NULL, NULL) == -1)
return -1;
if (!FD_ISSET(sock, &writefds))
return -1;
n = write(sock, buf + total, bytesleft);
if (n <= 0)
break;
total += n;
bytesleft -= n;
}
return (n == -1) ? -1 : total;
}
/* Receive a frame sent by HAProxy. It returns -1 if an error occurred,
* otherwise the number of read bytes.*/
static int
read_frame(int sock, struct worker *w)
{
uint32_t netint;
unsigned int framesz;
/* Read the frame size, on 4 bytes */
if (do_read(sock, &netint, sizeof(netint)) != 4) {
w->status_code = SPOE_FRM_ERR_IO;
return -1;
}
/* Check it against the max size */
framesz = ntohl(netint);
if (framesz > w->size) {
w->status_code = SPOE_FRM_ERR_TOO_BIG;
return -1;
}
/* Read the frame */
if (do_read(sock, w->buf, framesz) != framesz) {
w->status_code = SPOE_FRM_ERR_IO;
return -1;
}
w->len = framesz;
return framesz;
}
/* Send a frame to HAProxy. It returns -1 if an error occurred, otherwise the
* number of written bytes. */
static int
write_frame(int sock, struct worker *w)
{
uint32_t netint;
/* Write the frame size, on 4 bytes */
netint = htonl(w->len);
if (do_write(sock, &netint, sizeof(netint)) != 4) {
w->status_code = SPOE_FRM_ERR_IO;
return -1;
}
/* Write the frame */
if (do_write(sock, w->buf, w->len) != w->len) {
w->status_code = SPOE_FRM_ERR_IO;
return -1;
}
return w->len;
}
/* Encode a variable-length integer. This function never fails and returns the
* number of written bytes. */
static int
encode_spoe_varint(uint64_t i, char *buf)
{
int idx;
if (i < 240) {
buf[0] = (unsigned char)i;
return 1;
}
buf[0] = (unsigned char)i | 240;
i = (i - 240) >> 4;
for (idx = 1; i >= 128; ++idx) {
buf[idx] = (unsigned char)i | 128;
i = (i - 128) >> 7;
}
buf[idx++] = (unsigned char)i;
return idx;
}
/* Decode a varable-length integer. If the decoding fails, -1 is returned. This
* happens when the buffer's end in reached. On success, the number of read
* bytes is returned. */
static int
decode_spoe_varint(char *buf, char *end, uint64_t *i)
{
unsigned char *msg = (unsigned char *)buf;
int idx = 0;
if (msg > (unsigned char *)end)
return -1;
if (msg[0] < 240) {
*i = msg[0];
return 1;
}
*i = msg[0];
do {
++idx;
if (msg+idx > (unsigned char *)end)
return -1;
*i += (uint64_t)msg[idx] << (4 + 7 * (idx-1));
} while (msg[idx] >= 128);
return (idx + 1);
}
/* Encode a string. The string will be prefix by its length, encoded as a
* variable-length integer. This function never fails and returns the number of
* written bytes. */
static int
encode_spoe_string(const char *str, size_t len, char *dst)
{
int idx = 0;
if (!len) {
dst[0] = 0;
return 1;
}
idx += encode_spoe_varint(len, dst);
memcpy(dst+idx, str, len);
return (idx + len);
}
/* Decode a string. Its length is decoded first as a variable-length integer. If
* it succeeds, and if the string length is valid, the begin of the string is
* saved in <*str>, its length is saved in <*len> and the total numbre of bytes
* read is returned. If an error occurred, -1 is returned and <*str> remains
* NULL. */
static int
decode_spoe_string(char *buf, char *end, char **str, uint64_t *len)
{
int r, idx = 0;
*str = NULL;
*len = 0;
if ((r = decode_spoe_varint(buf, end, len)) == -1)
goto error;
idx += r;
if (buf + idx + *len > end)
goto error;
*str = buf+idx;
return (idx + *len);
error:
return -1;
}
/* Skip a typed data. If an error occurred, -1 is returned, otherwise the number
* of bytes read is returned. A types data is composed of a type (1 byte) and
* corresponding data:
* - boolean: non additional data (0 bytes)
* - integers: a variable-length integer (see decode_spoe_varint)
* - ipv4: 4 bytes
* - ipv6: 16 bytes
* - binary and string: a buffer prefixed by its size, a variable-length
* integer (see decode_spoe_string) */
static int
skip_spoe_data(char *frame, char *end)
{
uint64_t sz = 0;
int r, idx = 0;
if (frame > end)
return -1;
switch (frame[idx++] & SPOE_DATA_T_MASK) {
case SPOE_DATA_T_BOOL:
idx++;
break;
case SPOE_DATA_T_INT32:
case SPOE_DATA_T_INT64:
case SPOE_DATA_T_UINT32:
case SPOE_DATA_T_UINT64:
if ((r = decode_spoe_varint(frame+idx, end, &sz)) == -1)
return -1;
idx += r;
break;
case SPOE_DATA_T_IPV4:
idx += 4;
break;
case SPOE_DATA_T_IPV6:
idx += 16;
break;
case SPOE_DATA_T_STR:
case SPOE_DATA_T_BIN:
if ((r = decode_spoe_varint(frame+idx, end, &sz)) == -1)
return -1;
idx += r + sz;
break;
}
if (frame+idx > end)
return -1;
return idx;
}
/* Decode a typed data. If an error occurred, -1 is returned, otherwise the
* number of read bytes is returned. See skip_spoe_data for details. */
static int
decode_spoe_data(char *frame, char *end, struct spoe_data *data)
{
uint64_t sz = 0;
int type, r, idx = 0;
if (frame > end)
return -1;
type = frame[idx++];
data->type = (type & SPOE_DATA_T_MASK);
switch (data->type) {
case SPOE_DATA_T_BOOL:
data->u.boolean = ((type & SPOE_DATA_FL_TRUE) == SPOE_DATA_FL_TRUE);
break;
case SPOE_DATA_T_INT32:
if ((r = decode_spoe_varint(frame+idx, end, &sz)) == -1)
return -1;
data->u.sint32 = sz;
idx += r;
break;
case SPOE_DATA_T_INT64:
if ((r = decode_spoe_varint(frame+idx, end, &sz)) == -1)
return -1;
data->u.uint32 = sz;
idx += r;
break;
case SPOE_DATA_T_UINT32:
if ((r = decode_spoe_varint(frame+idx, end, &sz)) == -1)
return -1;
data->u.sint64 = sz;
idx += r;
break;
case SPOE_DATA_T_UINT64:
if ((r = decode_spoe_varint(frame+idx, end, &sz)) == -1)
return -1;
data->u.uint64 = sz;
idx += r;
break;
case SPOE_DATA_T_IPV4:
if (frame+idx+4 > end)
return -1;
memcpy(&data->u.ipv4, frame+idx, 4);
idx += 4;
break;
case SPOE_DATA_T_IPV6:
if (frame+idx+16 > end)
return -1;
memcpy(&data->u.ipv6, frame+idx, 16);
idx += 16;
break;
case SPOE_DATA_T_STR:
if ((r = decode_spoe_varint(frame+idx, end, &sz)) == -1)
return -1;
idx += r;
if (frame+idx+sz > end)
return -1;
data->u.buffer.str = frame+idx;
data->u.buffer.len = sz;
idx += sz;
break;
case SPOE_DATA_T_BIN:
if ((r = decode_spoe_varint(frame+idx, end, &sz)) == -1)
return -1;
idx += r;
if (frame+idx+sz > end)
return -1;
data->u.buffer.str = frame+idx;
data->u.buffer.len = sz;
idx += sz;
break;
default:
break;
}
if (frame+idx > end)
return -1;
return idx;
}
/* Check the protocol version. It returns -1 if an error occurred, the number of
* read bytes otherwise. */
static int
check_proto_version(struct worker *w, int idx)
{
char *str;
uint64_t sz;
/* Get the list of all supported versions by HAProxy */
if ((w->buf[idx++] & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR) {
w->status_code = SPOE_FRM_ERR_INVALID;
return -1;
}
idx += decode_spoe_string(w->buf+idx, w->buf+w->len, &str, &sz);
if (str == NULL) {
w->status_code = SPOE_FRM_ERR_INVALID;
return -1;
}
/* TODO: Find the right version in supported ones */
return idx;
}
/* Check max frame size value. It returns -1 if an error occurred, the number of
* read bytes otherwise. */
static int
check_max_frame_size(struct worker *w, int idx)
{
uint64_t sz;
int type, i;
/* Get the max-frame-size value of HAProxy */
type = w->buf[idx++];
if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
(type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
(type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
(type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64) {
w->status_code = SPOE_FRM_ERR_INVALID;
return -1;
}
if ((i = decode_spoe_varint(w->buf+idx, w->buf+w->len, &sz)) == -1) {
w->status_code = SPOE_FRM_ERR_INVALID;
return -1;
}
idx += i;
/* Keep the lower value */
if (sz < w->size)
w->size = sz;
return idx;
}
/* Check healthcheck value. It returns -1 if an error occurred, the number of
* read bytes otherwise. */
static int
check_healthcheck(struct worker *w, int idx)
{
int type;
/* Get the "healthcheck" value of HAProxy */
type = w->buf[idx++];
if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_BOOL) {
w->status_code = SPOE_FRM_ERR_INVALID;
return -1;
}
w->healthcheck = ((type & SPOE_DATA_FL_TRUE) == SPOE_DATA_FL_TRUE);
return idx;
}
/* Decode a HELLO frame received from HAProxy. It returns -1 if an error
* occurred, 0 if the frame must be skipped, otherwise the number of read
* bytes. */
static int
handle_hahello(struct worker *w)
{
char *end = w->buf+w->len;
int i, idx = 0;
/* Check frame type */
if (w->buf[idx++] != SPOE_FRM_T_HAPROXY_HELLO)
goto skip;
/* Skip flags */
idx += 4;
/* stream-id and frame-id must be cleared */
if (w->buf[idx] != 0 || w->buf[idx+1] != 0) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
idx += 2;
/* Loop on K/V items */
while (idx < w->len) {
char *str;
uint64_t sz;
/* Decode the item name */
idx += decode_spoe_string(w->buf+idx, end, &str, &sz);
if (str == NULL) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
/* Check "supported-versions" K/V item */
if (!memcmp(str, "supported-versions", sz)) {
if ((i = check_proto_version(w, idx)) == -1)
goto error;
idx = i;
}
/* Check "max-frame-size" K/V item "*/
else if (!memcmp(str, "max-frame-size", sz)) {
if ((i = check_max_frame_size(w, idx)) == -1)
goto error;
idx = i;
}
/* Check "healthcheck" K/V item "*/
else if (!memcmp(str, "healthcheck", sz)) {
if ((i = check_healthcheck(w, idx)) == -1)
goto error;
idx = i;
}
/* Skip "capabilities" K/V item for now */
else {
/* Silently ignore unknown item */
if ((i = skip_spoe_data(w->buf+idx, end)) == -1) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
idx += i;
}
}
return idx;
skip:
return 0;
error:
return -1;
}
/* Decode a DISCONNECT frame received from HAProxy. It returns -1 if an error
* occurred, 0 if the frame must be skipped, otherwise the number of read
* bytes. */
static int
handle_hadiscon(struct worker *w)
{
char *end = w->buf+w->len;
int i, idx = 0;
/* Check frame type */
if (w->buf[idx++] != SPOE_FRM_T_HAPROXY_DISCON)
goto skip;
/* Skip flags */
idx += 4;
/* stream-id and frame-id must be cleared */
if (w->buf[idx] != 0 || w->buf[idx+1] != 0) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
idx += 2;
/* Loop on K/V items */
while (idx < w->len) {
char *str;
uint64_t sz;
/* Decode item key */
idx += decode_spoe_string(w->buf+idx, end, &str, &sz);
if (str == NULL) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
/* Silently ignore unknown item */
if ((i = skip_spoe_data(w->buf+idx, end)) == -1) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
idx += i;
}
w->status_code = SPOE_FRM_ERR_NONE;
return idx;
skip:
return 0;
error:
return -1;
}
/* Encode a ACK frame to send it to HAProxy. It returns -1 if an error occurred,
* the number of written bytes otherwise. */
static void prepare_agentack(struct worker *w)
{
unsigned int flags = 0;
w->ack_len = 0;
/* Frame type */
w->ack[w->ack_len++] = SPOE_FRM_T_AGENT_ACK;
/* Set flags */
flags |= htonl(SPOE_FRM_FL_FIN);
memcpy(w->ack + w->ack_len, &flags, 4);
w->ack_len += 4;
/* Set stream-id and frame-id for ACK frames */
w->ack_len += encode_spoe_varint(w->stream_id, w->ack + w->ack_len);
w->ack_len += encode_spoe_varint(w->frame_id, w->ack + w->ack_len);
}
static inline
int set_var_name(struct worker *w, const char *name, int name_len, unsigned char scope)
{
w->ack[w->ack_len++] = SPOE_ACT_T_SET_VAR; /* Action type */
w->ack[w->ack_len++] = 3; /* Number of args */
w->ack[w->ack_len++] = scope; /* Arg 1: the scope */
w->ack_len += encode_spoe_string(name, name_len, w->ack+w->ack_len); /* Arg 2: variable name */
return 1;
}
int set_var_null(struct worker *w,
const char *name, int name_len,
unsigned char scope)
{
if (!set_var_name(w, name, name_len, scope))
return 0;
w->ack[w->ack_len++] = SPOE_DATA_T_NULL;
return 1;
}
int set_var_bool(struct worker *w,
const char *name, int name_len,
unsigned char scope, bool value)
{
if (!set_var_name(w, name, name_len, scope))
return 0;
w->ack[w->ack_len++] = SPOE_DATA_T_BOOL | (!!value << 4);
return 1;
}
static inline
int set_var_int(struct worker *w,
const char *name, int name_len,
unsigned char scope, int type, uint64_t value)
{
if (!set_var_name(w, name, name_len, scope))
return 0;
w->ack[w->ack_len++] = SPOE_DATA_T_UINT32;
w->ack_len += encode_spoe_varint(value, w->ack+w->ack_len); /* Arg 3: variable value */
return 1;
}
int set_var_uint32(struct worker *w,
const char *name, int name_len,
unsigned char scope, uint32_t value)
{
return set_var_int(w, name, name_len, scope, SPOE_DATA_T_UINT32, value);
}
int set_var_int32(struct worker *w,
const char *name, int name_len,
unsigned char scope, int32_t value)
{
return set_var_int(w, name, name_len, scope, SPOE_DATA_T_INT32, value);
}
int set_var_uint64(struct worker *w,
const char *name, int name_len,
unsigned char scope, uint64_t value)
{
return set_var_int(w, name, name_len, scope, SPOE_DATA_T_INT32, value);
}
int set_var_int64(struct worker *w,
const char *name, int name_len,
unsigned char scope, int64_t value)
{
return set_var_int(w, name, name_len, scope, SPOE_DATA_T_INT32, value);
}
int set_var_ipv4(struct worker *w,
const char *name, int name_len,
unsigned char scope,
struct in_addr *ipv4)
{
if (!set_var_name(w, name, name_len, scope))
return 0;
w->ack[w->ack_len++] = SPOE_DATA_T_IPV4;
memcpy(w->ack+w->ack_len, ipv4, 4);
w->ack_len += 4;
return 1;
}
int set_var_ipv6(struct worker *w,
const char *name, int name_len,
unsigned char scope,
struct in6_addr *ipv6)
{
if (!set_var_name(w, name, name_len, scope))
return 0;
w->ack[w->ack_len++] = SPOE_DATA_T_IPV6;
memcpy(w->ack+w->ack_len, ipv6, 16);
w->ack_len += 16;
return 1;
}
static inline
int set_var_buf(struct worker *w,
const char *name, int name_len,
unsigned char scope, int type,
const char *str, int str_len)
{
if (!set_var_name(w, name, name_len, scope))
return 0;
w->ack[w->ack_len++] = type;
w->ack_len += encode_spoe_string(str, str_len, w->ack+w->ack_len);
return 1;
}
int set_var_string(struct worker *w,
const char *name, int name_len,
unsigned char scope,
const char *str, int strlen)
{
return set_var_buf(w, name, name_len, scope, SPOE_DATA_T_STR, str, strlen);
}
int set_var_bin(struct worker *w,
const char *name, int name_len,
unsigned char scope,
const char *str, int strlen)
{
return set_var_buf(w, name, name_len, scope, SPOE_DATA_T_BIN, str, strlen);
}
/* This function is a little bit ugly,
* TODO: improve the response without copying the buffer
*/
static int commit_agentack(struct worker *w)
{
memcpy(w->buf, w->ack, w->ack_len);
w->len = w->ack_len;
return 1;
}
/* Decode a NOTIFY frame received from HAProxy. It returns -1 if an error
* occurred, 0 if the frame must be skipped, otherwise the number of read
* bytes. */
static int
handle_hanotify(struct worker *w)
{
char *end = w->buf+w->len;
uint64_t stream_id, frame_id;
int nbargs, i, idx = 0;
int index;
struct spoe_kv args[256];
uint64_t length;
struct ps_message *msg;
/* Check frame type */
if (w->buf[idx++] != SPOE_FRM_T_HAPROXY_NOTIFY)
goto skip;
/* Skip flags */
idx += 4;
/* Read the stream-id */
if ((i = decode_spoe_varint(w->buf+idx, end, &stream_id)) == -1) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
idx += i;
/* Read the frame-id */
if ((i = decode_spoe_varint(w->buf+idx, end, &frame_id)) == -1) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
idx += i;
w->stream_id = (unsigned int)stream_id;
w->frame_id = (unsigned int)frame_id;
DEBUG("Notify frame received: stream-id=%u - frame-id=%u",
w->stream_id, w->frame_id);
/* Prepare ack, if the processing fails the ack will be cancelled */
prepare_agentack(w);
/* Loop on messages */
while (idx < w->len) {
char *str;
uint64_t sz;
/* Decode the message name */
idx += decode_spoe_string(w->buf+idx, end, &str, &sz);
if (str == NULL) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
DEBUG(" Message '%.*s' received", (int)sz, str);
/* Decode all SPOE data */
nbargs = (unsigned char)w->buf[idx++];
for (index = 0; index < nbargs; index++) {
/* Read the key name */
if ((i = decode_spoe_string(w->buf+idx, end,
&args[index].name.str,
&length)) == -1) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
if (length > INT_MAX) {
w->status_code = SPOE_FRM_ERR_TOO_BIG;
goto error;
}
args[index].name.len = length;
idx += i;
/* Read the value */
memset(&args[index].value, 0, sizeof(args[index].value));
if ((i = decode_spoe_data(w->buf+idx, end, &args[index].value)) == -1) {
w->status_code = SPOE_FRM_ERR_INVALID;
goto error;
}
idx += i;
}
/* Lookup for existing bindings. If no existing message
* where found, does nothing.
*/
for (msg = ps_messages; msg; msg = msg->next)
if (sz == strlen(msg->name) && strncmp(str, msg->name, sz) == 0)
break;
if (msg == NULL || msg->ps->exec_message == NULL) {
DEBUG(" Message '%.*s' have no bindings registered", (int)sz, str);
continue;
}
/* Process the message */
msg->ps->exec_message(w, msg->ref, nbargs, args);
}
return idx;
skip:
return 0;
error:
return -1;
}
/* Encode a HELLO frame to send it to HAProxy. It returns -1 if an error
* occurred, the number of written bytes otherwise. */
static int
prepare_agenthello(struct worker *w)
{
int idx = 0;
unsigned int flags = 0;
/* Frame Type */
w->buf[idx++] = SPOE_FRM_T_AGENT_HELLO;
/* Set flags */
flags |= htonl(SPOE_FRM_FL_FIN);
memcpy(w->buf+idx, &flags, 4);
idx += 4;
/* No stream-id and frame-id for HELLO frames */
w->buf[idx++] = 0;
w->buf[idx++] = 0;
/* "version" K/V item */
idx += encode_spoe_string("version", 7, w->buf+idx);
w->buf[idx++] = SPOE_DATA_T_STR;
idx += encode_spoe_string(SPOP_VERSION, SLEN(SPOP_VERSION), w->buf+idx);
/* "max-frame-size" K/V item */
idx += encode_spoe_string("max-frame-size", 14, w->buf+idx);
w->buf[idx++] = SPOE_DATA_T_UINT32;
idx += encode_spoe_varint(w->size, w->buf+idx);
/* "capabilities" K/V item */
idx += encode_spoe_string("capabilities", 12, w->buf+idx);
w->buf[idx++] = SPOE_DATA_T_STR;
idx += encode_spoe_string(SPOA_CAPABILITIES, SLEN(SPOA_CAPABILITIES), w->buf+idx);
w->len = idx;
return idx;
}
/* Encode a DISCONNECT frame to send it to HAProxy. It returns -1 if an error
* occurred, the number of written bytes otherwise. */
static int
prepare_agentdicon(struct worker *w)
{
const char *reason;
int rlen, idx = 0;
unsigned int flags = 0;
if (w->status_code >= SPOE_FRM_ERRS)
w->status_code = SPOE_FRM_ERR_UNKNOWN;
reason = spoe_frm_err_reasons[w->status_code];
rlen = strlen(reason);
/* Frame type */
w->buf[idx++] = SPOE_FRM_T_AGENT_DISCON;
/* Set flags */
flags |= htonl(SPOE_FRM_FL_FIN);
memcpy(w->buf+idx, &flags, 4);
idx += 4;
/* No stream-id and frame-id for DISCONNECT frames */
w->buf[idx++] = 0;
w->buf[idx++] = 0;
/* There are 2 mandatory items: "status-code" and "message" */
/* "status-code" K/V item */
idx += encode_spoe_string("status-code", 11, w->buf+idx);
w->buf[idx++] = SPOE_DATA_T_UINT32;
idx += encode_spoe_varint(w->status_code, w->buf+idx);
/* "message" K/V item */
idx += encode_spoe_string("message", 7, w->buf+idx);
w->buf[idx++] = SPOE_DATA_T_STR;
idx += encode_spoe_string(reason, rlen, w->buf+idx);
w->len = idx;
return idx;
}
static int
hello_handshake(int sock, struct worker *w)
{
if (read_frame(sock, w) < 0) {
LOG("Failed to read Haproxy HELLO frame");
goto error;
}
if (handle_hahello(w) < 0) {
LOG("Failed to handle Haproxy HELLO frame");
goto error;
}
if (prepare_agenthello(w) < 0) {
LOG("Failed to prepare Agent HELLO frame");
goto error;
}
if (write_frame(sock, w) < 0) {
LOG("Failed to write Agent frame");
goto error;
}
DEBUG("Hello handshake done: version=%s - max-frame-size=%u - healthcheck=%s",
SPOP_VERSION, w->size, (w->healthcheck ? "true" : "false"));
return 0;
error:
return -1;
}
static int
notify_ack_roundtip(int sock, struct worker *w)
{
if (read_frame(sock, w) < 0) {
LOG("Failed to read Haproxy NOTIFY frame");
goto error_or_quit;
}
if (handle_hadiscon(w) != 0) {
if (w->status_code != SPOE_FRM_ERR_NONE)
LOG("Failed to handle Haproxy DISCONNECT frame");
DEBUG("Disconnect frame received: reason=%s",
spoe_frm_err_reasons[w->status_code]);
goto error_or_quit;
}
if (handle_hanotify(w) < 0) {
LOG("Failed to handle Haproxy NOTIFY frame");
goto error_or_quit;
}
if (commit_agentack(w) < 0) {
LOG("Failed to prepare Agent ACK frame");
goto error_or_quit;
}
if (write_frame(sock, w) < 0) {
LOG("Failed to write Agent ACK frame");
goto error_or_quit;
}
DEBUG("Ack frame sent: stream-id=%u - frame-id=%u",
w->stream_id, w->frame_id);
return 0;
error_or_quit:
return -1;
}
static void *
spoa_worker(void *data)
{
struct worker w;
struct sockaddr_in client;
int *info = (int *)data;
int csock, lsock = info[0];
struct ps *ps;
int i;
int len;
signal(SIGPIPE, SIG_IGN);
pthread_setspecific(worker_id, &info[1]);
/* Init registered processors */
for (ps = ps_list; ps != NULL; ps = ps->next)
ps->init_worker(&w);
/* Load files */
for (i = 0; i < nfiles; i++) {
len = strlen(files[i]);
for (ps = ps_list; ps != NULL; ps = ps->next)
if (strcmp(files[i] + len - strlen(ps->ext), ps->ext) == 0)
break;
if (ps == NULL) {
LOG("Can't load file \"%s\"\n", files[i]);
goto out;
}
if (!ps->load_file(&w, files[i]))
goto out;
}
while (1) {
socklen_t sz = sizeof(client);
if ((csock = accept(lsock, (struct sockaddr *)&client, &sz)) < 0) {
LOG("Failed to accept client connection: %m");
goto out;
}
memset(&w, 0, sizeof(w));
w.id = info[1];
w.size = MAX_FRAME_SIZE;
DEBUG("New connection from HAProxy accepted");
if (hello_handshake(csock, &w) < 0)
goto disconnect;
if (w.healthcheck == true)
goto close;
while (1) {
if (notify_ack_roundtip(csock, &w) < 0)
break;
}
disconnect:
if (w.status_code == SPOE_FRM_ERR_IO) {
LOG("Close the client socket because of I/O errors");
goto close;
}
if (prepare_agentdicon(&w) < 0) {
LOG("Failed to prepare Agent DISCONNECT frame");
goto close;
}
if (write_frame(csock, &w) < 0) {
LOG("Failed to write Agent DISCONNECT frame");
goto close;
}
DEBUG("Disconnect frame sent: reason=%s",
spoe_frm_err_reasons[w.status_code]);
close:
close(csock);
}
out:
free(info);
#if 0
pthread_exit(NULL);
#endif
return NULL;
}
int process_create(pid_t *pid, void *(*ps)(void *), void *data)
{
if (debug) {
ps(data);
exit(EXIT_SUCCESS);
}
*pid = fork();
if (*pid == -1)
return -1;
if (*pid > 0)
return 0;
ps(data);
return 0;
}
static void
usage(char *prog)
{
fprintf(stderr, "Usage: %s [-h] [-d] [-p <port>] [-n <num-workers>] -f <file>\n", prog);
fprintf(stderr, " -h Print this message\n");
fprintf(stderr, " -d Enable the debug mode\n");
fprintf(stderr, " -p <port> Specify the port to listen on (default: 12345)\n");
fprintf(stderr, " -n <num-workers> Specify the number of workers (default: 5)\n");
fprintf(stderr, " -f <file> Specify the file whoch contains the processing code.\n");
fprintf(stderr, " This argument can specified more than once.\n");
}
int
main(int argc, char **argv)
{
#if 0
pthread_t *ts = NULL;
#endif
pid_t *pids;
struct sockaddr_in server;
int i, sock, opt, nbworkers, port;
int status;
nbworkers = NUM_WORKERS;
port = DEFAULT_PORT;
while ((opt = getopt(argc, argv, "hdn:p:f:")) != -1) {
switch (opt) {
case 'h':
usage(argv[0]);
return EXIT_SUCCESS;
case 'd':
debug = true;
break;
case 'n':
nbworkers = atoi(optarg);
break;
case 'p':
port = atoi(optarg);
break;
case 'f':
add_file(optarg);
break;
default:
usage(argv[0]);
return EXIT_FAILURE;
}
}
if (nbworkers <= 0) {
fprintf(stderr, "%s: Invalid number of workers '%d'\n",
argv[0], nbworkers);
goto error;
}
if (port <= 0) {
fprintf(stderr, "%s: Invalid port '%d'\n", argv[0], port);
goto error;
}
if((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
fprintf(stderr, "Failed creating socket: %m\n");
goto error;
}
setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (int []){1}, sizeof(int));
setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (int []){1}, sizeof(int));
memset(&server, 0, sizeof(server));
server.sin_family = AF_INET;
server.sin_addr.s_addr = INADDR_ANY;
server.sin_port = htons(port);
if (bind(sock, (struct sockaddr *)&server, sizeof(server)) < 0) {
fprintf(stderr, "Failed to bind the socket: %m\n");
goto error;
}
if (listen(sock , 10) < 0) {
fprintf(stderr, "Failed to listen on the socket: %m\n");
goto error;
}
fprintf(stderr, "SPOA is listening on port %d\n", port);
pthread_key_create(&worker_id, NULL);
/* Initialise the server in thread mode. This code is commented
* out and not deleted, because later I expect to work with
* process ansd threads. This first version just support processes.
*/
#if 0
ts = calloc(nbworkers, sizeof(*ts));
for (i = 0; i < nbworkers; i++) {
int *info = calloc(2, sizeof(*info));
info[0] = sock;
info[1] = i+1;
if (pthread_create(&ts[i], NULL, spoa_worker, info) < 0) {
fprintf(stderr, "Failed to create thread %d: %m\n", i+1);
goto error;
}
fprintf(stderr, "SPOA worker %02d started\n", i+1);
}
for (i = 0; i < nbworkers; i++) {
pthread_join(ts[i], NULL);
fprintf(stderr, "SPOA worker %02d stopped\n", i+1);
}
free(ts);
#endif
/* Start processes */
pids = calloc(nbworkers, sizeof(*pids));
if (!pids) {
fprintf(stderr, "Out of memory error\n");
goto error;
}
for (i = 0; i < nbworkers; i++) {
int *info = calloc(2, sizeof(*info));
info[0] = sock;
info[1] = i+1;
if (process_create(&pids[i], spoa_worker, info) == -1) {
fprintf(stderr, "SPOA worker %02d started\n", i+1);
goto error;
}
fprintf(stderr, "SPOA worker %02d started\n", i+1);
}
for (i = 0; i < nbworkers; i++) {
waitpid(pids[0], &status, 0);
fprintf(stderr, "SPOA worker %02d stopped\n", i+1);
}
close(sock);
pthread_key_delete(worker_id);
return EXIT_SUCCESS;
error:
return EXIT_FAILURE;
}
Reference in New Issue View Git Blame Copy Permalink