diff --git a/plugins/out_http/http.c b/plugins/out_http/http.c index 7df3657e294..5fef56f1bb2 100644 --- a/plugins/out_http/http.c +++ b/plugins/out_http/http.c @@ -76,6 +76,20 @@ static int cb_http_init(struct flb_output_instance *ins, return 0; } +static const char *http_method_name(int method) +{ + switch (method) { + case FLB_HTTP_POST: + return "POST"; + case FLB_HTTP_PUT: + return "PUT"; + case FLB_HTTP_PATCH: + return "PATCH"; + } + + return "UNKNOWN"; +} + /* header names and values must not contain line breaks */ static int header_is_valid(const char *str, size_t len) { @@ -647,7 +661,7 @@ static int send_all_requests(struct flb_out_http *ctx, if (body_found && headers_found) { flb_plg_trace(ctx->ins, "sending record %zu via %s", record_count++, - ctx->http_method == FLB_HTTP_POST ? "POST" : "PUT"); + http_method_name(ctx->http_method)); ret = http_request(ctx, body, body_size, event_chunk->tag, flb_sds_len(event_chunk->tag), headers); } @@ -902,7 +916,7 @@ static struct flb_config_map config_map[] = { { FLB_CONFIG_MAP_STR, "http_method", "POST", 0, FLB_FALSE, 0, - "Specify the HTTP method to use. Supported methods are POST and PUT" + "Specify the HTTP method to use. Supported methods are POST, PUT and PATCH" }, /* Gelf Properties */ diff --git a/plugins/out_http/http_conf.c b/plugins/out_http/http_conf.c index 47d39c9fc91..7731b4ee025 100644 --- a/plugins/out_http/http_conf.c +++ b/plugins/out_http/http_conf.c @@ -293,7 +293,9 @@ struct flb_out_http *flb_http_conf_create(struct flb_output_instance *ins, } else { flb_plg_error(ctx->ins, "invalid compress option '%s'", tmp); - flb_free(ctx); + ctx->u = upstream; + ctx->uri = uri; + flb_http_conf_destroy(ctx); return NULL; } } @@ -308,9 +310,17 @@ struct flb_out_http *flb_http_conf_create(struct flb_output_instance *ins, else if (strcasecmp(tmp, "PUT") == 0) { ctx->http_method = FLB_HTTP_PUT; } + else if (strcasecmp(tmp, "PATCH") == 0) { + ctx->http_method = FLB_HTTP_PATCH; + } else { - flb_plg_error(ctx->ins, "invalid http_method option '%s'. Supported methods are POST and PUT", tmp); - flb_free(ctx); + flb_plg_error(ctx->ins, + "invalid http_method option '%s'. " + "Supported methods are POST, PUT and PATCH", + tmp); + ctx->u = upstream; + ctx->uri = uri; + flb_http_conf_destroy(ctx); return NULL; } } diff --git a/tests/runtime/out_http.c b/tests/runtime/out_http.c index 615c3f9b427..3c6bf52c3f5 100644 --- a/tests/runtime/out_http.c +++ b/tests/runtime/out_http.c @@ -18,9 +18,19 @@ * limitations under the License. */ +#include +#include +#ifndef _WIN32 +#include +#include +#include +#include +#endif + #include #include #include +#include #include #include #include @@ -1201,6 +1211,276 @@ void flb_test_in_http() test_ctx_destroy(ctx); } +#ifndef _WIN32 +struct patch_method_mock_server { + flb_sockfd_t listen_fd; + int port; + int stop; + pthread_t thread; + pthread_mutex_t lock; + char request_method[16]; +}; + +static int patch_method_mock_server_stopping(struct patch_method_mock_server *server) +{ + int stop; + + pthread_mutex_lock(&server->lock); + stop = server->stop; + pthread_mutex_unlock(&server->lock); + + return stop; +} + +static void patch_method_mock_server_set_method(struct patch_method_mock_server *server, + const char *method) +{ + pthread_mutex_lock(&server->lock); + strncpy(server->request_method, method, + sizeof(server->request_method) - 1); + pthread_mutex_unlock(&server->lock); +} + +static void patch_method_mock_server_get_method(struct patch_method_mock_server *server, + char *out, size_t size) +{ + pthread_mutex_lock(&server->lock); + strncpy(out, server->request_method, size - 1); + out[size - 1] = '\0'; + pthread_mutex_unlock(&server->lock); +} + +static void *patch_method_mock_server_thread(void *data) +{ + struct patch_method_mock_server *server; + flb_sockfd_t conn_fd; + ssize_t bytes; + char request[4096]; + const char response[] = + "HTTP/1.1 200 OK\r\n" + "Content-Length: 0\r\n" + "Connection: close\r\n" + "\r\n"; + + server = data; + + while (!patch_method_mock_server_stopping(server)) { + conn_fd = accept(server->listen_fd, NULL, NULL); + if (patch_method_mock_server_stopping(server)) { + if (conn_fd >= 0) { + flb_socket_close(conn_fd); + } + break; + } + if (conn_fd < 0) { + flb_time_msleep(10); + continue; + } + + bytes = recv(conn_fd, request, sizeof(request) - 1, 0); + if (bytes > 0) { + request[bytes] = '\0'; + if (strncmp(request, "PATCH ", 6) == 0) { + patch_method_mock_server_set_method(server, "PATCH"); + } + else if (strncmp(request, "POST ", 5) == 0) { + patch_method_mock_server_set_method(server, "POST"); + } + else if (strncmp(request, "PUT ", 4) == 0) { + patch_method_mock_server_set_method(server, "PUT"); + } + } + + send(conn_fd, response, strlen(response), 0); + flb_socket_close(conn_fd); + break; + } + + return NULL; +} + +static int patch_method_mock_server_start(struct patch_method_mock_server *server) +{ + int on = 1; + struct sockaddr_in addr; + socklen_t len; + + memset(server, 0, sizeof(struct patch_method_mock_server)); + if (pthread_mutex_init(&server->lock, NULL) != 0) { + server->listen_fd = -1; + return -1; + } + + server->listen_fd = socket(AF_INET, SOCK_STREAM, 0); + if (server->listen_fd < 0) { + pthread_mutex_destroy(&server->lock); + return -1; + } + + setsockopt(server->listen_fd, SOL_SOCKET, SO_REUSEADDR, (const char *) &on, + sizeof(on)); + + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = 0; + + if (bind(server->listen_fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) { + flb_socket_close(server->listen_fd); + server->listen_fd = -1; + pthread_mutex_destroy(&server->lock); + return -1; + } + + len = sizeof(addr); + if (getsockname(server->listen_fd, (struct sockaddr *) &addr, &len) < 0) { + flb_socket_close(server->listen_fd); + server->listen_fd = -1; + pthread_mutex_destroy(&server->lock); + return -1; + } + + server->port = ntohs(addr.sin_port); + + if (listen(server->listen_fd, 4) < 0) { + flb_socket_close(server->listen_fd); + server->listen_fd = -1; + pthread_mutex_destroy(&server->lock); + return -1; + } + + server->stop = 0; + if (pthread_create(&server->thread, NULL, patch_method_mock_server_thread, + server) != 0) { + flb_socket_close(server->listen_fd); + server->listen_fd = -1; + pthread_mutex_destroy(&server->lock); + return -1; + } + + return 0; +} + +static void patch_method_mock_server_stop(struct patch_method_mock_server *server) +{ + if (server->listen_fd < 0) { + return; + } + + flb_sockfd_t wake_fd; + struct sockaddr_in addr; + + pthread_mutex_lock(&server->lock); + server->stop = 1; + pthread_mutex_unlock(&server->lock); + + /* wake the listener in case no client ever connected */ + wake_fd = socket(AF_INET, SOCK_STREAM, 0); + if (wake_fd >= 0) { + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = htons(server->port); + connect(wake_fd, (struct sockaddr *) &addr, sizeof(addr)); + flb_socket_close(wake_fd); + } + shutdown(server->listen_fd, SHUT_RDWR); + + pthread_join(server->thread, NULL); + flb_socket_close(server->listen_fd); + server->listen_fd = -1; + pthread_mutex_destroy(&server->lock); +} + +void flb_test_http_method_patch(void) +{ + struct patch_method_mock_server server; + struct test_ctx *ctx; + int ret; + int attempts; + char port[16]; + char method[16]; + char *buf = "[1, {\"msg\":\"hello world\"}]"; + size_t size = strlen(buf); + + memset(&server, 0, sizeof(server)); + server.listen_fd = -1; + + if (!TEST_CHECK(patch_method_mock_server_start(&server) == 0)) { + TEST_MSG("failed to start mock HTTP server"); + return; + } + + snprintf(port, sizeof(port), "%d", server.port); + + clear_output_num(); + + ctx = test_ctx_create(); + if (!TEST_CHECK(ctx != NULL)) { + patch_method_mock_server_stop(&server); + TEST_MSG("test_ctx_create failed"); + return; + } + + ret = flb_output_set(ctx->flb, ctx->o_ffd, + "match", "*", + "format", "json", + "host", "127.0.0.1", + "port", port, + "http_method", "patch", + NULL); + TEST_CHECK(ret == 0); + + ret = flb_start(ctx->flb); + TEST_CHECK(ret == 0); + + ret = flb_lib_push(ctx->flb, ctx->i_ffd, (char *) buf, size); + TEST_CHECK(ret >= 0); + + for (attempts = 0; attempts < 50; attempts++) { + patch_method_mock_server_get_method(&server, method, sizeof(method)); + if (method[0] != '\0') { + break; + } + flb_time_msleep(100); + } + + patch_method_mock_server_stop(&server); + + /* server thread is joined, safe to read */ + if (!TEST_CHECK(strcmp(server.request_method, "PATCH") == 0)) { + TEST_MSG("expected PATCH request, got '%s'", server.request_method); + } + + test_ctx_destroy(ctx); +} + +void flb_test_http_method_invalid(void) +{ + struct test_ctx *ctx; + int ret; + + ctx = test_ctx_create(); + if (!TEST_CHECK(ctx != NULL)) { + TEST_MSG("test_ctx_create failed"); + return; + } + + ret = flb_output_set(ctx->flb, ctx->o_ffd, + "match", "*", + "format", "json", + "http_method", "DELETE", + NULL); + TEST_CHECK(ret == 0); + + ret = flb_start(ctx->flb); + TEST_CHECK(ret != 0); + + flb_destroy(ctx->flb); + flb_free(ctx); +} +#endif + /* Test list */ TEST_LIST = { {"format_msgpack" , flb_test_format_msgpack}, @@ -1218,5 +1498,9 @@ TEST_LIST = { {"json_date_format_iso8601" , flb_test_json_date_format_iso8601}, {"json_date_format_java_sql_timestamp" , flb_test_json_date_format_java_sql_timestamp}, {"in_http", flb_test_in_http}, +#ifndef _WIN32 + {"http_method_patch", flb_test_http_method_patch}, + {"http_method_invalid", flb_test_http_method_invalid}, +#endif {NULL, NULL} };