New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

filterx pubsub message #456

Open

bshifter wants to merge 4 commits into axoflow:main from bshifter:fx-pubsub

+553 −21

Member

bshifter commented Jan 15, 2025

Add pubsub_message() function to FilterX and extend Pub/Sub destination with protovar

This PR introduces new functionality for managing Google Pub/Sub messages in syslog-ng:

pubsub_message() Function:

Adds a FilterX representation of the Google Pub/Sub message Protobuf descriptor.
Enables users to define and manipulate Pub/Sub message content directly within the FilterX scripting module (source-side).

protovar Option in Pub/Sub Destination:

Extends the Pub/Sub destination module with the protovar option.
Allows referencing the FilterX-defined Pub/Sub Protobuf variable for setting the outgoing message payload.
Together, these changes provide greater flexibility and control over Pub/Sub destination messages, supporting advanced customization directly from FilterX.

example:

filterx {
    # Adding data and attributes 
    $x = pubsub_message($MESSAGE, {"foo":{"bar":"baz"}});
};

log {
    destination {
        google-pubsub-grpc(
            ...
            proto_var($x)
        );
    };
};

Contributor

github-actions bot commented Jan 15, 2025 •

edited

Loading

This Pull Request introduces config grammar changes

axoflow/7a663104e8c0914fc63cb0d3da350f988dda1f50 -> bshifter/fx-pubsub

--- a/destination
+++ b/destination

 google-pubsub-grpc(
+    proto-var(
+        <template-content>
+        <template-reference>
+    )
 )

bshifter force-pushed the fx-pubsub branch 3 times, most recently from f7510c9 to d15fbc7 Compare

January 16, 2025 10:27

alltilla self-requested a review

January 16, 2025 12:45

bshifter force-pushed the fx-pubsub branch from d15fbc7 to 318dbda Compare

January 17, 2025 11:15

bshifter added 4 commits

January 17, 2025 12:16


          grpc: fix grpc sources dir

ac8ab67

Signed-off-by: shifter <[email protected]>


          filterx: add google pubsub protobuf object wrapper

c7ba621

The implementation follows the existing patterns and designs of the OpenTelemetry (OTel) filterx modules for consistency.

Signed-off-by: shifter <[email protected]>


          grpc: extends pubsub destination with protovar option

e894513

The protovar option in the Pub/Sub destination allows users to directly access the Protobuf byte representation created by a FilterX function on the source side.

The Protobuf serialized variable is used to set the Pub/Sub message's content.
This enables users to fully control and manage the outgoing messages of the Pub/Sub destination from within FilterX.
By leveraging this option, users can customize the message payload dynamically, based on their specific requirements.
The data() and attributes() options cannot be used with protovar(). These options are mutually exclusive.

Signed-off-by: shifter <[email protected]>


          filterx: add light test for pubsub_message

471948f

Signed-off-by: shifter <[email protected]>

bshifter force-pushed the fx-pubsub branch from 318dbda to 471948f Compare

January 17, 2025 11:16

alltilla reviewed

View reviewed changes

modules/grpc/pubsub/filterx/object-pubsub-message.cpp

Comment on lines +2 to +3

		* Copyright (c) 2024 Axoflow
		* Copyright (c) 2023 shifter

Member

alltilla Jan 17, 2025

Please use the current year.

modules/grpc/pubsub/filterx/object-pubsub-message.cpp

+                          {
+                            std::string data_cpp(filterx_string_get_value_ref(data_arg, NULL));
+                            std::map<std::string, std::string> attributes_cpp;
+                            if (!filterx_dict_iter(attributes_arg, _build_map, static_cast<gpointer>(&attributes_cpp)))

Member

alltilla Jan 17, 2025

We iterate through the dict once here, than through the map created here once again the the ctor. I believe there is opportunity for optimization.

modules/grpc/pubsub/filterx/object-pubsub-message.cpp

+                        FilterXObject *data_arg = filterx_ref_unwrap_ro(data);
+                        FilterXObject *attributes = args[1];
+                        FilterXObject *attributes_arg = filterx_ref_unwrap_ro(attributes);
+                        if (filterx_object_is_type(data_arg, &FILTERX_TYPE_NAME(string)) &&

Member

alltilla Jan 17, 2025

There is an interesting question here: what types should we support here? The PubSub gRPC proto allows us to send any bytes data, not just string, so I believe we should support everything here. The question then arises: how should we format each type? I think for string, bytes and protobuf get_value() can be used for performance optimization, and filterx_object_marshal() for every other type would bring us far.

modules/grpc/pubsub/filterx/object-pubsub-message.cpp

+                  }
+                catch (const std::runtime_error &e)
+                  {
+                    msg_error("FilterX: Failed to create Pubsup Message object", evt_tag_str("error", e.what()));

Member

alltilla Jan 17, 2025

typo: Pubsub -> PubSub or Pub/Sub

modules/grpc/pubsub/filterx/object-pubsub-message.cpp

Comment on lines +180 to +183

+                if (!filterx_object_repr(key, key_str))
+                  return FALSE;
+                if (!filterx_object_repr(val, val_str))
+                  return FALSE;

Member

alltilla Jan 17, 2025

In the majority of the time these will already be strings, and repr() copies strings even. For strings we could call get_value(), to optimize out the copy, for the rest repr() is perfect.

modules/grpc/pubsub/filterx/object-pubsub.h

Comment on lines +2 to +3

		* Copyright (c) 2024 Axoflow
		* Copyright (c) 2023 shifter

Member

alltilla Jan 17, 2025

Please use recent dates.

modules/grpc/pubsub/filterx/object-pubsub.h

+              FILTERX_DECLARE_TYPE(pubsub_message);
+              static inline void
+              pubsub_filterx_objects_global_init(void)

Member

alltilla Jan 17, 2025

Shouldn't this be called in google_pubsub_grpc_module_init()?

modules/grpc/pubsub/pubsub-dest-worker.cpp

-                    buf_slice = this->format_template(attribute.value, msg, buf, NULL, this->super->super.seq_num);
-                    attributes->insert({attribute.name, buf_slice.str});
-                    message_bytes += buf_slice.len;
+                    if (log_template_is_trivial(owner_->protovar))

Member

alltilla Jan 17, 2025

IMO this check should go to the setter and we should refuse the value if it is not trivial.

modules/grpc/pubsub/pubsub-dest-worker.cpp

Comment on lines +138 to +139

		std::cout << "Key: \"" << key << "\", Length: " << key.length() << "\n";
		std::cout << "Value: \"" << value << "\", Length: " << value.length() << "\n";

Member

alltilla Jan 17, 2025

We should remove these debug messages.

modules/grpc/pubsub/pubsub-dest-worker.cpp

-                auto attributes = message->mutable_attributes();
-                for (const auto &attribute : owner_->attributes)
+                if (owner_->protovar)

Member

alltilla Jan 17, 2025

[optional]

I think this is a good candidate for factoring out to some smaller functions.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet