Event subscribing bug

[YujiOshima]

With regard to event subscribe, topics that is not subscribed are also delivered from the server.
For example, / timer / sec / 10 and / timer / sec / 100 are also subscribed if I subscribe / timer / sec / 1.
This is because go-radix does a string-level search that does not treat / specially.

[chungers]

@YujiOshima - To fix this the plugin RPC client can verify the topic names to enforce the proper topics -- they are just types.Path which can be checked against the topics returned by the List() method of the SPI. This way, the core pkg/broker package's radix tree index doesn't have to have knowledge of the schema of a topic which is enforced as a / separated path by the Plugin. What do you think?

[YujiOshima]

@chungers In that way you can exclude illegal topic subscribe, but if the proper topic provided by event plugin partially overlaps, the method is not enough.
For example, when the timer publishes /timer/sec/1 and /timer/sec/10 as described above.
I think that it is not practical to impose restrictions on all plugins that all topics can not overlap in part.
But I agree your opinion that radix tree index shouldn't have to have knowledge of the schema.
To solve it, How about limiting topic's specification with reference to mqtt?

• A subscription to A/# is a subscription to the topic A and all topics beneath A(may be using . instead of # in this case).
• Not using #, subscribe only exact match topic.

If you can agree this rule, I will show you some codes for discuss.

[chungers]

@YujiOshima - I like how you referenced MQTT. I had originally thought about using it but I couldn't find a good embeddable broker. We basically need just something smaller (and not as featureful) for just the plugins on a single host, over unix sockets. For intra-cluster and inter-cluster pubsub beyond a single host, the plan is to provide simple 'repeaters' that is both a subscriber and publisher to republish events using MQTT or NATSD which the user can pick. An MQTT one would make a nice PR ;)

Anyway, I hope you agree with me about adding validation checks in an interceptor and return http error code (eg. 404) for bad topics. This keeps layering clean and we can adopt whatever syntax that makes sense for specifying topics and we won't have to do any filters and checks at message delivery time.

Using your example for topic naming, the timer plugin publishes to two topics

• timer/sec/1
• timer/sec/10

These are the rules:

• Subscribe totimer/sec/1 ==> matches timer/sec/1
• Subscribe totimer/sec/10 ==> matches timer/sec/10
• Subscribe to timer/sec ==> matches timer/sec. It's valid. A publisher may come on later and publish to this topic.
• Subscribe to timer/sec/ ==> matches timer/sec/1 and timer/sec/10 and timer/sec. The pkg/broker sees a string timer/sec/ which will delimit the topic properly for the radix tree.
• Subscribe to timer/ ==> matches everything under timer.

So timer/sec/ is like timer/sec/# as you proposed. But it's intentionally more like a filesystem syntax (e.g. ls -al global/cluster1/timer/sec/ lists all topics, tail global/cluster1/time/sec/ subscribes to all underneath).

What do you think?

[YujiOshima]

@chungers Yes, topic validation is nice as it reduce workload of broker and make simple its process.
I agree almost all your subscribing rules.
One point, I am not sure when you subscribe to timer/sec/, whether timer/sec topic should be matched or only below topics except timer/sec.
I think both is OK, so we go ahead with your plan for the time being.

Inter host is the next step that I am strongly interested in.
Repeater sounds good. I think repeater should be implemented in Infrakit manager or something, not each plugins. So we should design carefully for that it won't be performance bottleneck.
There are also many tasks (API/rpc, not dir scan plugin discovery, etc) for it.
I would love to discuss about inter host support.

At first, I will make spi/api conform above rules.

[chungers]

@YujiOshima - Cool. I am glad the current design looks reasonable to you.

I think it'd be very nice to have a repeater that can subscribe to some topics and then acts as a client to a MQTT server and publish. The repeater should not be crammed into the plugins. Instead, it can be a very small binary where the user can start up like:

infrakit-event-relay -mqtt <url> --relay infrakit/topic/1=mqtt/TOPIC2 --relay infrakit//topic/3=mqtt/TOPIC3 ...

The binary would then embed the MQTT client from the Eclipse project or something. Then the user only needs to set up a single MQTT server for the entire cluster and infrakit plugins can publish to that. This would make a very interesting case for running infrakit on every node in the cluster now... may be for health check /monitoring / whatever.

Let me know if you would rather try something like this.... I can fix this issue with adding the validation via an interceptor of ServeHTTP. That way you can work on the relay/repeater to help prove out the concept.

[YujiOshima]

@chungers Will infrakit-event-relay be implemented as a plugin?
I agree it be plugable to maintain interchangeability. Currently it is reasonable to use mqtt, but I’m not sure that mqtt is the best over the future. :p

It is considered that there is a problem with activating infrakit-event-relay by allowing the user to specify options for the following reasons.

• Restarting infrakit-event-relay is necessary when a topic that you want to publish is generated later.
• It is very troublesome to specify all topics to publish.

Therefore, how about the following rules?

• No option : add prefix /plugin-name/ and use Use the topic of event plugin as it is
• set option —delimiter=/:. : infrakit-event-relay interpret delimiter of event topic is / and replace it as .. Then add prefix /plugin-name/.
• set option —topic infrakit/topic/1=mqtt/TOPIC2 : infrakit/topic/1 event topic publish as mqtt/TOPIC2 topic and add prefix /plugin-name/.

What do you think?

[chungers]

@YujiOshima

I think the message relay should be a separate binary. It can expose its own REST api for adding new subscriptions. This could be the first of an 'app' that is built on top of infrakit infrastructure. An infrakit 'app' is an application that can talk to the infrakit plugins using the plugin api but itself listens on a port and offers a REST api for operations like adding subscriptions, etc. So if the user decides it's useful to stream all the infrakit events to something like ELK or MQTT or NASTD or Kafka then this would be the module to use.

If MQTT isn't a good long term option, what do you think would be a good message broker to start? It'd have meet the requirements of

• easy set up
• small footprint
• easy to integrate with 3rd party dashboards

If you are interested in help building this relay we can iterate on that simple REST api for adding new subscriptions. What do you think?

[YujiOshima]

@chungers

Although mqtt is concerned about acces control, redundancy etc., I think that it is not at the stage of discussing them.
First of all, I think that it is important to show by implementation, so adopting mqtt is good.
In the future, if you wish to see acces control, load balancing subscription, you may need to consider amqp, kafka, http based things, but if you pluggable it would not be a problem.
Indeed mqtt has enough requirements in many usecase and that support of mqtt will be necessary also in the future.
anyway, I will create a PoC using mqtt. Open a new issue and create a PR.