Implement UDP socket target

Implements [0] a new target to send datagrams to UDP unix socket.
This has useful application with low latency to trigger
external events.

[0] #96

CMakeLists.txt

@@ -100,6 +100,7 @@ add_subdirectory( ut )
 add_subdirectory( tools/matching-helper )
 add_subdirectory( tools/arashpartow-helper )
 add_subdirectory( tools/h2client )
+add_subdirectory( tools/udp-server )
 
 ###########
 # Install #

Dockerfile

@@ -22,6 +22,7 @@ COPY --from=builder /code/build/${build_type}/bin/h2agent /opt/
 COPY --from=builder /code/build/${build_type}/bin/h2client /opt/
 COPY --from=builder /code/build/${build_type}/bin/matching-helper /opt/
 COPY --from=builder /code/build/${build_type}/bin/arashpartow-helper /opt/
+COPY --from=builder /code/build/${build_type}/bin/udp-server /opt/
 
 # We add curl & jq for helpers.src
 # Ubuntu has bash already installed, but vim is missing

Dockerfile.training

@@ -34,6 +34,7 @@ RUN ln -s /opt/h2agent
 RUN ln -s /opt/h2client
 RUN ln -s /opt/matching-helper
 RUN ln -s /opt/arashpartow-helper
+RUN ln -s /opt/udp-server
 
 ENTRYPOINT ["sleep", "infinity"]
 CMD []

README.md

@@ -76,6 +76,8 @@ The option `--auto` builds the <u>builder image</u> (`--builder-image`) , then t
   $> docker run --rm -it --network=host --entrypoint "/opt/matching-helper" ghcr.io/testillano/h2agent:latest --help
   -or-
   $> docker run --rm -it --network=host --entrypoint "/opt/arashpartow-helper" ghcr.io/testillano/h2agent:latest --help
+  -or-
+  $> docker run --rm -it --entrypoint "/opt/udp-server" ghcr.io/testillano/h2agent:latest --help
   ```
 
 * Run within `kubernetes` deployment: corresponding `helm charts` are normally packaged into releases. This is described in ["how it is delivered"](#How-it-is-delivered) section, but in summary, you could do the following:
@@ -803,7 +805,7 @@ Expression: 404 == 404
 Result: 1
 ```
 
-## Execution of h2client helper utility
+## Execution of h2client utility
 
 This utility could be useful to test simple HTTP/2 requests.
 
@@ -873,6 +875,49 @@ Uri: http://localhost:8000/book/8472098362
  Response headers: [date: Sun, 27 Nov 2022 18:58:32 GMT]
 ```
 
+## Execution of udp-server utility
+
+This utility could be useful to test UDP messages sent by `h2agent` (`udpSocket.*` target).
+You can also use netcat in bash, to generate messages easily:
+
+```bash
+echo -n "<message here>" | nc -u -q0 -w1 -U /tmp/my_unix_socket
+```
+
+### Command line
+
+You may take a look to `udp-server` command line by just typing the build path, for example for `Release` target using native executable:
+
+```bash
+$> ./build/Release/bin/udp-server --help
+Usage: udp-server [options]
+
+Options:
+
+--path <value>
+  UDP unix socket path
+
+--print-each <value>
+  Print messages each specific amount (must be positive). Defaults to 1.
+
+[-h|--help]
+  This help.
+
+Examples:
+   udp-server --path "/tmp/my_unix_socket"
+```
+
+Execution example:
+
+```bash
+$> ./build/Release/bin/udp-server --path "/tmp/my_unix_socket"
+
+Path: /tmp/my_unix_socket
+Print each: 1 message(s)
+
+Waiting for messages ([sequence] <message>) ...
+```
+
 ## Execution with TLS support
 
 `H2agent` server mock supports `SSL/TLS`. You may use helpers located under `tools/ssl` to create server key and certificate files:
@@ -1706,7 +1751,7 @@ Defines the response behavior for an incoming request matching some basic condit
           },
           "target": {
             "type": "string",
-            "pattern": "^response\\.body\\.(string$|hexstring$)|^response\\.body\\.json\\.(object$|object\\..+|jsonstring$|jsonstring\\..+|string$|string\\..+|integer$|integer\\..+|unsigned$|unsigned\\..+|float$|float\\..+|boolean$|boolean\\..+)|^response\\.(header\\..+|statusCode|delayMs)$|^(var|globalVar|serverEvent)\\..+|^outState(\\.(POST|GET|PUT|DELETE|HEAD)(\\..+)?)?$|^txtFile\\..+|^binFile\\..+|^break$"
+            "pattern": "^response\\.body\\.(string$|hexstring$)|^response\\.body\\.json\\.(object$|object\\..+|jsonstring$|jsonstring\\..+|string$|string\\..+|integer$|integer\\..+|unsigned$|unsigned\\..+|float$|float\\..+|boolean$|boolean\\..+)|^response\\.(header\\..+|statusCode|delayMs)$|^(var|globalVar|serverEvent)\\..+|^outState(\\.(POST|GET|PUT|DELETE|HEAD)(\\..+)?)?$|^txtFile\\..+|^binFile\\..+|^udpSocket\\..+|^break$"
           }
         },
         "additionalProperties" : {
@@ -2027,8 +2072,12 @@ The **target** of information is classified after parsing the following possible
 
 - txtFile.`<path>` *[string]*: dumps source (as string) over text file with the path provided. The path can be relative (to the execution directory) or absolute, and **admits variables substitution**. Note that paths to missing directories will fail to open (the process does not create tree hierarchy). It is considered long term file (file is closed 1 second after last write, by default) when a constant path is configured, because this is normally used for specific log files. On the other hand, when any substitution may took place in the path provided (it has variables in the form `@{varname}`) it is considered as a dynamic name, so understood as short term file (file is opened, written and closed without delay, by default). **Note:** you can force short term type inserting a variable, for example with empty value: `txtFile./path/to/short-term-file.txt@{empty}`. Delays in microseconds are configurable on process startup. Check  [command line](#Command-line) for `--long-term-files-close-delay-usecs` and `--short-term-files-close-delay-usecs` options.
 
+  This target can also be used to write named pipes (previously created: `mkfifo /tmp/mypipe && chmod 0666 /tmp/mypipe`), with the following restriction: writes must close the file descriptor everytime, so long/short term delays for close operations must be zero depending on which of them applies: variable paths zeroes the delay by default, but constant ones shall be zeroed too by command-line (`--long-term-files-close-delay-usecs 0`). Just like with regular UNIX pipes (`|`), when the writer closes, the pipe is torn down, so fast operations writting named pipes could provoke data looses (some writes missed). In that case, it is more recommended to use UDP unix sockets target (`udpSocket./tmp/myunixsocket`).
+
 - binFile.`<path>` *[string]*: same as `txtFile` but writting binary data.
 
+- udpSocket.`<path>[.<milliseconds delay>]` *[string]*: sends source (as string) towards the UDP unix socket with the path provided, with an optional delay in milliseconds (if path contains dots, an unexpected delay can be configured and same could happen for the path, so in that case, you must force delay specification by mean adding `.0` or any valid value). The path can be relative (to the execution directory) or absolute, and **admits variables substitution**. UDP is a transport layer protocol in the TCP/IP suite, which provides a simple, connectionless, and unreliable communication service. It is a lightweight protocol that does not guarantee the delivery or order of data packets. Instead, it allows applications to send individual datagrams (data packets) to other hosts over the network without establishing a connection first. UDP is often used where low latency is crucial. In `h2agent` is useful to signal external applications to do associated tasks sharing specific data for the transactions processed. Use `./tools/udp-server` program to play with it.
+
 - serverEvent.`<server event address in query parameters format>`: this target is always used in conjunction with `eraser` source acting as an alternative purge method to the purge `outState`. The main difference is that states-driven purge method acts over processed events key (`method` and `uri` for the provision in which the purge state is planned), so not all the test scenarios may be covered with that constraint if they need to remove events registered for different transactions. In this case, event addressing is defined by request *method* (`requestMethod`), *URI* (`requestUri`), and events *number* (`eventNumber`): events number *path* (`eventPath`) is not accepted, as this operation just remove specific events or whole history, like REST API for server-data deletion:
 
   - *requestMethod*: any supported method (*POST*, *GET*, *PUT*, *DELETE*, *HEAD*). Mandatory.
@@ -3048,7 +3097,7 @@ Client provisions are a fundamental part of the client mode configuration. Unlik
           },
           "target": {
             "type": "string",
-            "pattern": "^request\\.body\\.(string$|hexstring$)|^request\\.body\\.json\\.(object$|object\\..+|jsonstring$|jsonstring\\..+|string$|string\\..+|integer$|integer\\..+|unsigned$|unsigned\\..+|float$|float\\..+|boolean$|boolean\\..+)|^request\\.(header\\..+|delayMs|timeoutMs)$|^(var|globalVar|clientEvent)\\..+|^outState$|^txtFile\\..+|^binFile\\..+"
+            "pattern": "^request\\.body\\.(string$|hexstring$)|^request\\.body\\.json\\.(object$|object\\..+|jsonstring$|jsonstring\\..+|string$|string\\..+|integer$|integer\\..+|unsigned$|unsigned\\..+|float$|float\\..+|boolean$|boolean\\..+)|^request\\.(header\\..+|delayMs|timeoutMs)$|^(var|globalVar|clientEvent)\\..+|^outState$|^txtFile\\..+|^binFile\\..+|^udpSocket\\..+"
           }
         },
         "additionalProperties" : {
@@ -3071,7 +3120,7 @@ Client provisions are a fundamental part of the client mode configuration. Unlik
           },
           "target": {
             "type": "string",
-            "pattern": "^(var|globalVar|clientEvent)\\..+|^outState$|^txtFile\\..+|^binFile\\..+|^break$"
+            "pattern": "^(var|globalVar|clientEvent)\\..+|^outState$|^txtFile\\..+|^binFile\\..+|^udpSocket\\..+|^break$"
           }
         },
         "additionalProperties" : {
@@ -3843,6 +3892,12 @@ FileSystem_observed_operations_total{operation="write"} 100000
 FileSystem_observed_operations_total{operation="close"} 1
 FileSystem_observed_operations_total{operation="empty"} 0
 FileSystem_observed_operations_total{operation="open"} 1
+# TYPE UDPSocket_observed_operations_total counter
+UDPSocket_observed_operations_total{operation="open",success="false"} 0
+UDPSocket_observed_operations_total{operation="delayedWrite"} 0
+UDPSocket_observed_operations_total{operation="instantWrite"} 0
+UDPSocket_observed_operations_total{operation="open"} 0
+UDPSocket_observed_operations_total{operation="write"} 0
 # HELP AdminHttp2Server_responses_delay_seconds_gauge Http2 message responses delay gauge (seconds) in AdminHttp2Server
 # TYPE AdminHttp2Server_responses_delay_seconds_gauge gauge
 AdminHttp2Server_responses_delay_seconds_gauge 7.2e-05
@@ -3907,6 +3962,7 @@ So, metrics implemented could be divided in two categories, **counters** and **g
 - DELETE requests
 - HEAD requests
 - <other> requests
+- File system and Unix sockets operations
 - Error-condition requests (POST/GET/PUT/DELETE/HEAD/other)
 
 #### Gauges and histograms

ct/src/conftest.py

@@ -38,6 +38,7 @@
 ADMIN_CONFIGURATION_URI = ADMIN_URI_PREFIX + 'configuration'
 ADMIN_GLOBAL_VARIABLE_URI = ADMIN_URI_PREFIX + 'global-variable'
 ADMIN_FILES_URI = ADMIN_URI_PREFIX + 'files'
+ADMIN_SOCKETS_URI = ADMIN_URI_PREFIX + 'udp-sockets'
 ADMIN_LOGGING_URI = ADMIN_URI_PREFIX + 'logging'
 ADMIN_SERVER_CONFIGURATION_URI = ADMIN_URI_PREFIX + 'server/configuration'
 ADMIN_SERVER_MATCHING_URI = ADMIN_URI_PREFIX + 'server-matching'
@@ -638,6 +639,36 @@ def send(content, responseBodyRef = VALID_CLIENT_PROVISION__RESPONSE_BODY, respo
 }
 '''
 
+SOCKET_MANAGER_PROVISION='''
+{
+  "requestMethod": "GET",
+  "requestUri":"/app/v1/foo/bar",
+  "responseCode": 200,
+  "transform": [
+    {
+      "source": "value./tmp/my_unix_socket1",
+      "target": "var.file1"
+    },
+    {
+      "source": "value./tmp/my_unix_socket2",
+      "target": "var.file2"
+    },
+    {
+      "source": "value.0",
+      "target": "var.delayms"
+    },
+    {
+      "source": "value.hello",
+      "target": "udpSocket.@{file1}.@{delayms}"
+    },
+    {
+      "source": "value.world",
+      "target": "udpSocket.@{file2}.@{delayms}"
+    }
+  ]
+}
+'''
+
 GLOBAL_VARIABLE_PROVISION_TEMPLATE_GVARCREATED_GVARREMOVED_GVARANSWERED='''
 {{
   "requestMethod":"POST",

ct/src/files_operation/fo_test.py

@@ -1,6 +1,6 @@
 import pytest
 import json
-import time
+#import time
 from conftest import ADMIN_FILES_URI, string2dict, FILE_MANAGER_PROVISION
 
 
@@ -26,7 +26,7 @@ def test_001_i_want_to_get_process_files(h2ac_admin, admin_server_provision, h2a
 #  # Wait 2 seconds (long-term file closes in 1 second by default)
 #  time.sleep(2)
 #
-  # Check file
+  # Check files json
   response = h2ac_admin.get(ADMIN_FILES_URI)
   responseBodyRef = [{ "bytes":5, "path": "/tmp/example.txt", "state": "closed" }]
   h2ac_admin.assert_response__status_body_headers(response, 200, responseBodyRef)

ct/src/sockets_operation/so2_test.py

@@ -0,0 +1,26 @@
+import pytest
+import json
+from conftest import ADMIN_SOCKETS_URI, string2dict, SOCKET_MANAGER_PROVISION
+
+
+@pytest.mark.admin
+def test_001_i_want_to_get_process_sockets(h2ac_admin, admin_server_provision, h2ac_traffic):
+
+  # Provision
+  admin_server_provision(string2dict(SOCKET_MANAGER_PROVISION))
+
+  # Check file before traffic: skipped because the test could re-run
+  #response = h2ac_admin.get(ADMIN_SOCKETS_URI)
+  #assert response["status"] == 204
+
+  # Send GET
+  response = h2ac_traffic.get("/app/v1/foo/bar")
+  h2ac_admin.assert_response__status_body_headers(response, 200, "")
+
+  # Check sockets json
+  response = h2ac_admin.get(ADMIN_SOCKETS_URI)
+  responseBodyRef = [{ "socket":0, "path": "/tmp/my_unix_socket2" }, { "socket":0, "path": "/tmp/my_unix_socket1" }]
+  responseBodyRef[0]["socket"] = response["body"][0]["socket"]
+  responseBodyRef[1]["socket"] = response["body"][1]["socket"]
+  h2ac_admin.assert_response__status_body_headers(response, 200, responseBodyRef)
+

src/http2/MyAdminHttp2Server.cpp

@@ -53,6 +53,7 @@ SOFTWARE.
 #include <Configuration.hpp>
 #include <GlobalVariable.hpp>
 #include <FileManager.hpp>
+#include <SocketManager.hpp>
 #include <functions.hpp>
 
 
@@ -570,6 +571,10 @@ void MyAdminHttp2Server::receiveGET(const std::string &uri, const std::string &p
         responseBody = getFileManager()->asJsonString();
         statusCode = ((responseBody == "[]") ? ert::http2comm::ResponseCode::NO_CONTENT:ert::http2comm::ResponseCode::OK); // 204 or 200
     }
+    else if (pathSuffix == "udp-sockets") {
+        responseBody = getSocketManager()->asJsonString();
+        statusCode = ((responseBody == "[]") ? ert::http2comm::ResponseCode::NO_CONTENT:ert::http2comm::ResponseCode::OK); // 204 or 200
+    }
     else if (pathSuffix == "logging") {
         responseBody = ert::tracing::Logger::levelAsString(ert::tracing::Logger::getLevel());
         headers.emplace("content-type", nghttp2::asio_http2::header_value{"text/html"});

src/http2/MyAdminHttp2Server.hpp

@@ -52,6 +52,7 @@ namespace model
 class Configuration;
 class GlobalVariable;
 class FileManager;
+class SocketManager;
 class AdminData;
 class MockServerData;
 class MockClientData;
@@ -131,6 +132,13 @@ class MyAdminHttp2Server: public ert::http2comm::Http2Server
         return common_resources_.FileManagerPtr;
     }
 
+    void setSocketManager(model::SocketManager *p) {
+        common_resources_.SocketManagerPtr = p;
+    }
+    model::SocketManager *getSocketManager() const {
+        return common_resources_.SocketManagerPtr;
+    }
+
     void setMetricsData(ert::metrics::Metrics *metrics, const ert::metrics::bucket_boundaries_t &responseDelaySecondsHistogramBucketBoundaries, const ert::metrics::bucket_boundaries_t &messageSizeBytesHistogramBucketBoundaries) {
         common_resources_.MetricsPtr = metrics;
         common_resources_.ResponseDelaySecondsHistogramBucketBoundaries = responseDelaySecondsHistogramBucketBoundaries;

src/http2/MyTrafficHttp2Client.cpp

@@ -51,6 +51,7 @@ SOFTWARE.
 #include <Configuration.hpp>
 #include <GlobalVariable.hpp>
 #include <FileManager.hpp>
+#include <SocketManager.hpp>
 #include <functions.hpp>
 
 namespace h2agent

src/http2/MyTrafficHttp2Client.hpp

@@ -57,6 +57,7 @@ namespace model
 class Configuration;
 class GlobalVariable;
 class FileManager;
+class SocketManager;
 class AdminData;
 }
 

src/http2/MyTrafficHttp2Server.cpp

@@ -50,6 +50,7 @@ SOFTWARE.
 #include <Configuration.hpp>
 #include <GlobalVariable.hpp>
 #include <FileManager.hpp>
+#include <SocketManager.hpp>
 #include <functions.hpp>
 
 namespace h2agent

src/http2/MyTrafficHttp2Server.hpp

@@ -59,6 +59,7 @@ class MockClientData;
 class Configuration;
 class GlobalVariable;
 class FileManager;
+class SocketManager;
 class AdminData;
 }
 

src/jsonSchema/AdminSchemas.hpp

@@ -235,7 +235,7 @@ const nlohmann::json server_provision = R"(
           },
           "target": {
             "type": "string",
-            "pattern": "^response\\.body\\.(string$|hexstring$)|^response\\.body\\.json\\.(object$|object\\..+|jsonstring$|jsonstring\\..+|string$|string\\..+|integer$|integer\\..+|unsigned$|unsigned\\..+|float$|float\\..+|boolean$|boolean\\..+)|^response\\.(header\\..+|statusCode|delayMs)$|^(var|globalVar|serverEvent)\\..+|^outState(\\.(POST|GET|PUT|DELETE|HEAD)(\\..+)?)?$|^txtFile\\..+|^binFile\\..+|^break$"
+            "pattern": "^response\\.body\\.(string$|hexstring$)|^response\\.body\\.json\\.(object$|object\\..+|jsonstring$|jsonstring\\..+|string$|string\\..+|integer$|integer\\..+|unsigned$|unsigned\\..+|float$|float\\..+|boolean$|boolean\\..+)|^response\\.(header\\..+|statusCode|delayMs)$|^(var|globalVar|serverEvent)\\..+|^outState(\\.(POST|GET|PUT|DELETE|HEAD)(\\..+)?)?$|^txtFile\\..+|^binFile\\..+|^udpSocket\\..+|^break$"
           }
         },
         "additionalProperties" : {
@@ -426,7 +426,7 @@ const nlohmann::json client_provision = R"(
           },
           "target": {
             "type": "string",
-            "pattern": "^request\\.body\\.(string$|hexstring$)|^request\\.body\\.json\\.(object$|object\\..+|jsonstring$|jsonstring\\..+|string$|string\\..+|integer$|integer\\..+|unsigned$|unsigned\\..+|float$|float\\..+|boolean$|boolean\\..+)|^request\\.(header\\..+|delayMs|timeoutMs)$|^(var|globalVar|clientEvent)\\..+|^outState$|^txtFile\\..+|^binFile\\..+"
+            "pattern": "^request\\.body\\.(string$|hexstring$)|^request\\.body\\.json\\.(object$|object\\..+|jsonstring$|jsonstring\\..+|string$|string\\..+|integer$|integer\\..+|unsigned$|unsigned\\..+|float$|float\\..+|boolean$|boolean\\..+)|^request\\.(header\\..+|delayMs|timeoutMs)$|^(var|globalVar|clientEvent)\\..+|^outState$|^txtFile\\..+|^binFile\\..+|^udpSocket\\..+"
           }
         },
         "additionalProperties" : {
@@ -449,7 +449,7 @@ const nlohmann::json client_provision = R"(
           },
           "target": {
             "type": "string",
-            "pattern": "^(var|globalVar|clientEvent)\\..+|^outState$|^txtFile\\..+|^binFile\\..+|^break$"
+            "pattern": "^(var|globalVar|clientEvent)\\..+|^outState$|^txtFile\\..+|^binFile\\..+|^udpSocket\\..+|^break$"
           }
         },
         "additionalProperties" : {