Feature add subscriber service for creating/dropping subscriptions #4375

nathanielc · 2015-10-08T21:12:24Z

Basic idea is to allow for forks or subscriptions of the data stream to be defined. A subscription is tied to database and retention policy and will send all data received to the subscription endpoint. Currently the only implementation is a UDP protocol. We do not want to create back-pressure within InfluxDB hence UDP.

Usage:

The following query creates a subscription called sub0 that forks all incoming data to database mydb and policy default to the hosts localhost:9090 and remotehost:9090 in a round robin fashion. If you want to send all data to both local and remote hosts create two subscriptions.

CREATE SUBSCRIPTION sub0 ON mydb.default HOSTS 'localhost:9090', 'remotehost:9090' TYPE 'udp'

The following query drops the subscriptions:

DROP SUBSCRIPTION sub0 ON mydb.default

The following query lists all subscriptions:

SHOW SUBSCRIPTIONS

These subscriptions are stored in the raft meta data so the whole cluster will fork its data at each node respectively.

TODO:

~~Need to update the QL docs and write some basic dos on how to use subcriptions.~~
~~Detect new or deleted subscriptions and start/stop them.~~

nathanielc · 2015-10-08T21:19:19Z

tsdb/query_executor.go

@@ -193,6 +198,13 @@ func (q *QueryExecutor) ExecuteQuery(query *influxql.Query, database string, chu
 			case *influxql.ShowStatsStatement, *influxql.ShowDiagnosticsStatement:
 				// Send monitor-related queries to the monitor service.
 				res = q.MonitorStatementExecutor.ExecuteStatement(stmt)
+			case *influxql.CreateSubscriptionStatement, *influxql.DropSubscriptionStatement:


I need to trigger an event on the Subscriber service whenever a subscription is added or removed.

But I am realizing this is not the right place to tell the Subscriber to update, as only one server in a cluster will actually execute the query. Should I add a Subscriber to the MetaStore instead and have it perform this update after a save? Not entirely sure how to do this part. @pauldix

jwilder · 2015-10-09T02:41:06Z

Instead of:

CREATE SUBSCRIPTION sub0 ON mydb.default HOSTS 'localhost:9090', 'remotehost:9090' TYPE 'udp'

What about something like:

CREATE SUBSCRIPTION sub0 ON mydb.default DESTINATIONS 'udp://localhost:9090', 'udp://remotehost:9090'

I think just a TYPE 'udp' may not be flexible enough to add other types of endpoints in the future or have much flexibility to secure endpoints with authentication creds, etc..

URLs are pretty flexible for identifying the type of endpoint, it's location, specifying options, etc.. which might make it easier to expand without changing the query language each time.

For example:

CREATE SUBSCRIPTION sub0 ON mydb.default DESTINATIONS 'influxdb://datacenter2:8086?db=foo&u=&p=', 'riemann://remotehost: 5555', 'kafka://...,

otoolep · 2015-10-09T02:47:08Z

I like the URL idea.

jwilder · 2015-10-09T02:51:39Z

Also, this functionality seems like it should have many of the same capabilities as the hinted handoff (durability, queue messages for failed writes, batching, back-off, node removal, etc..). Any reason not to leverage some of that underlying infrastructure?

pauldix · 2015-10-09T14:33:02Z

@jwilder the goal for this is that it would never cause InfluxDB to back up writes. We didn't want to open up the possibility of causing InfluxDB to go down if the external system did.

Although it could be done in the future. Just doesn't make sense with UDP.

pauldix · 2015-10-09T14:33:30Z

I like the URL syntax that @jwilder suggested

nathanielc · 2015-10-09T14:37:53Z

+1 on the URL scheme.

Two questions:

How could we distinguish between multiple destinations and redundant destinations?

I might want to send data to any one of the destinations or to all of them. The use case for sending to just one is you have a backend that can scale and so you would like to balance across them, i.e. round robin balancing.

Something like this:

CREATE SUBSCRIPTION sub0 ON mydb.default DESTINATIONS ANY 'udp://localhost:9090', 'udp://remotehost:9090'

or

CREATE SUBSCRIPTION sub0 ON mydb.default DESTINATIONS ALL 'udp://localhost:9090', 'udp://remotehost:9090', 'kafka://....'

As for the HH idea. We were initially thinking that this should be very lightweight and if you drop data not a big deal. Rather keep the InfluxDB server happy then apply back pressure into it. And if you wanted to do something more complex then do it outside of the DB. @pauldix Thoughts? Reasons for using HH for subscriptions vs keeping it simple?

pauldix · 2015-10-09T14:39:50Z

@nathanielc I don't think the HH stuff makes sense with this initial case since it's UDP. I'd vote to keep it simple for now.

jwilder · 2015-10-09T15:21:23Z

@pauldix @nathanielc For UDP transport, I agree, HH-like functionality might not be that useful since you would never know if the receiver got the write. I was thinking more about if we did need a reliable transport for a subscription. For example, a s3 subscription that backs up all writes to S3 or a http endpoint where an application wants to ensure it always receives the post. For those use-cases, the subscriber transport would need to durably queue the writes and retry if the receiver failed. If would definitely need to shed load (drop new writes or expire old ones) if the queues filled up though. Seems like those concerns could be addressed in that specific subscriber implementation though.

nathanielc · 2015-10-09T15:26:15Z

@jwilder I like the idea of each subscriber handling the concerns of reliability.

nathanielc · 2015-10-09T16:49:49Z

influxql/INFLUXQL.md

+### CREATE SUBSCRIPTION
+
+```
+create_subscription_stmt = "CREATE SUBSCRIPTION" subscription_name "ON" db_name "." retention_policy "DESTINATIONS" ("ANY"|"ALL") host { "," host} .


Is this EBNF syntax correct? Not sure the ("ANY"|"ALL") part is.

Should it be like this?

create_subscription_stmt = "CREATE SUBSCRIPTION" subscription_name "ON" db_name "." retention_policy "DESTINATIONS" all_any host { "," host} . all_any = "ALL" | "ANY"

nathanielc · 2015-10-09T19:34:50Z

services/subscriber/service.go

+					}
+					writers[i] = w
+					tags := map[string]string{"database": se.db, "retention_policy": se.rp, "name": se.name, "mode": si.Mode}
+					statMaps[i] = influxdb.NewStatistics(dest, "subscriber", tags)


@otoolep Does this look good for the stats? I didn't add it to the UDP object since here we can tag the stats with more useful information.

What does dest look like?

its the URL specified in the query 'udp://host:port' should be unique to a given subscription. Does it need to be globally unique, as in all statMaps?

Looking at NewStatistics code now...

Ok, so key is global to all statMaps, I updated it to be unique now.

The best to be sure you're happy with the stats is fire up the system,
generate some activity and the execute the query 'SHOW STATS' at the CLI.
How does it look?

On Fri, Oct 9, 2015 at 1:04 PM, Nathaniel Cook notifications@github.com
wrote:

In services/subscriber/service.go
#4375 (comment):

return fmt.Errorf("unknown balance mode %q", si.Mode)

}

writers := make([]PointsWriter, len(si.Destinations))

statMaps := make([]*expvar.Map, len(writers))

for i, dest := range si.Destinations {

u, err := url.Parse(dest)

if err != nil {

return err

}

w, err := s.NewPointsWriter(*u)

if err != nil {

return err

}

writers[i] = w

tags := map[string]string{"database": se.db, "retention_policy": se.rp, "name": se.name, "mode": si.Mode}

statMaps[i] = influxdb.NewStatistics(dest, "subscriber", tags)

Ok, so key is global to all statMaps, I updated it to be unique now.

—
Reply to this email directly or view it on GitHub
https://github.com/influxdb/influxdb/pull/4375/files#r41672288.

Looks good:

name: subscriber ---------------- points_written write_failures 628 14 name: subscriber tags: database=telegraf, destination=udp://localhost:9093, mode=ALL, name=k, retention_policy=default points_written write_failures -------------- -------------- 342 14

Thanks

Don't you feel better already? :-)

Absolutely!
I am going to have to get monitoring setup on my side of things too soon. That was really easy and nice to use.

Remember the stats are going straight back into the _internal database, so
you can query there too, for historical insight.

On Fri, Oct 9, 2015 at 2:37 PM, Nathaniel Cook notifications@github.com
wrote:

In services/subscriber/service.go
#4375 (comment):

return fmt.Errorf("unknown balance mode %q", si.Mode)

}

writers := make([]PointsWriter, len(si.Destinations))

statMaps := make([]*expvar.Map, len(writers))

for i, dest := range si.Destinations {

u, err := url.Parse(dest)

if err != nil {

return err

}

w, err := s.NewPointsWriter(*u)

if err != nil {

return err

}

writers[i] = w

tags := map[string]string{"database": se.db, "retention_policy": se.rp, "name": se.name, "mode": si.Mode}

statMaps[i] = influxdb.NewStatistics(dest, "subscriber", tags)

Absolutely!
I am going to have to get monitoring setup on my side of things too soon.
That was really easy and nice to use.

—
Reply to this email directly or view it on GitHub
https://github.com/influxdb/influxdb/pull/4375/files#r41680738.

otoolep · 2015-10-13T21:33:33Z

meta/store_test.go

+	}
+}
+
+// Ensure the store can delete a continuous query.


And further on too.

otoolep · 2015-10-14T20:44:06Z

services/subscriber/service.go

+	var lastErr error
+	for range b.writers {
+		i := b.i
+		w := b.writers[i]


What is the point of the modulo stuff?

The module code rotates through destinations, otherwise each write would always try the first destination first and it would get a disproportionate amount of data.

otoolep · 2015-10-14T20:44:41Z

LGTM, one question.

jwilder · 2015-10-14T20:52:25Z

cmd/influxd/run/config_test.go

@@ -45,6 +45,9 @@ bind-address = ":4444"
 [monitoring]
 enabled = true

+[subscriber]
+enabled = false


Should probably use true since the default for a bool is false.

jwilder · 2015-10-14T21:03:46Z

LGTM. 👍

InfluxDB data stream.

Feature add subscriber service for creating/dropping subscriptions

manishjain002 · 2015-11-29T02:17:58Z

I have a use case, please see if this make sense in implementing in "Subscribe".
I have an application for monitoring of my sensors data that's getting logged. Before "Subscribe" , I had only one way, i.e. Pull the data from the DB server using "polling". But with subscription, I get data Pushed.
But there is a problem, I am getting all the data that is getting stored in the DB, I would be interested in filtering data. What I want is "selective Subscription".

beckettsean · 2015-11-30T20:31:37Z

@manishjain002 what you are describing is a feature request for Kapacitor. The repo for that project will be public as of tomorrow (Dec 1st) so I encourage you to open an issue there.

nathanielc added the 2 - Working label Oct 8, 2015

nathanielc force-pushed the subscriptions branch from d451a43 to f9563bf Compare October 8, 2015 21:15

nathanielc reviewed Oct 8, 2015
View reviewed changes

nathanielc force-pushed the subscriptions branch 4 times, most recently from 0594c2f to 32b914e Compare October 8, 2015 22:10

nathanielc mentioned this pull request Oct 8, 2015

Javascript error in Admin interface running 'SHOW CONTINUOUS QUERIES' #3820

Closed

nathanielc force-pushed the subscriptions branch from 32b914e to dc067df Compare October 9, 2015 16:46

nathanielc reviewed Oct 9, 2015
View reviewed changes

nathanielc force-pushed the subscriptions branch 2 times, most recently from a6f7640 to f79e1e5 Compare October 9, 2015 19:28

nathanielc reviewed Oct 9, 2015
View reviewed changes

nathanielc force-pushed the subscriptions branch 4 times, most recently from 262f5e9 to fb4486d Compare October 9, 2015 21:19

otoolep reviewed Oct 13, 2015
View reviewed changes

nathanielc force-pushed the subscriptions branch 6 times, most recently from 185e58a to d4b1dcf Compare October 14, 2015 19:43

otoolep reviewed Oct 14, 2015
View reviewed changes

nathanielc force-pushed the subscriptions branch from d4b1dcf to 5cededa Compare October 14, 2015 20:50

jwilder reviewed Oct 14, 2015
View reviewed changes

Adds subscriber service for creating/dropping subscriptions to the

8b31007

InfluxDB data stream.

nathanielc force-pushed the subscriptions branch from 5cededa to 8b31007 Compare October 14, 2015 21:24

nathanielc pushed a commit that referenced this pull request Oct 15, 2015

Merge pull request #4375 from influxdb/subscriptions

cb1aaa8

Feature add subscriber service for creating/dropping subscriptions

nathanielc merged commit cb1aaa8 into master Oct 15, 2015

nathanielc removed the 2 - Working label Oct 15, 2015

nathanielc deleted the subscriptions branch October 15, 2015 15:17

jackzampolin mentioned this pull request Dec 9, 2015

document SUBSCRIBE influxdata/docs.influxdata.com-ARCHIVE#38

Closed

skyrocknroll mentioned this pull request Dec 27, 2015

support for influxdb subscriptions riemann/riemann#635

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Feature add subscriber service for creating/dropping subscriptions #4375

Feature add subscriber service for creating/dropping subscriptions #4375

nathanielc commented Oct 8, 2015

nathanielc Oct 8, 2015

jwilder commented Oct 9, 2015

otoolep commented Oct 9, 2015

jwilder commented Oct 9, 2015

pauldix commented Oct 9, 2015

pauldix commented Oct 9, 2015

nathanielc commented Oct 9, 2015

pauldix commented Oct 9, 2015

jwilder commented Oct 9, 2015

nathanielc commented Oct 9, 2015

nathanielc Oct 9, 2015

nathanielc Oct 9, 2015

otoolep Oct 9, 2015

nathanielc Oct 9, 2015

nathanielc Oct 9, 2015

otoolep Oct 9, 2015

nathanielc Oct 9, 2015

otoolep Oct 9, 2015

nathanielc Oct 9, 2015

otoolep Oct 9, 2015

otoolep Oct 13, 2015

otoolep Oct 13, 2015

otoolep Oct 14, 2015

nathanielc Oct 14, 2015

otoolep commented Oct 14, 2015

jwilder Oct 14, 2015

jwilder commented Oct 14, 2015

manishjain002 commented Nov 29, 2015

beckettsean commented Nov 30, 2015

Feature add subscriber service for creating/dropping subscriptions #4375

Feature add subscriber service for creating/dropping subscriptions #4375

Conversation

nathanielc commented Oct 8, 2015

Choose a reason for hiding this comment

jwilder commented Oct 9, 2015

otoolep commented Oct 9, 2015

jwilder commented Oct 9, 2015

pauldix commented Oct 9, 2015

pauldix commented Oct 9, 2015

nathanielc commented Oct 9, 2015

pauldix commented Oct 9, 2015

jwilder commented Oct 9, 2015

nathanielc commented Oct 9, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

otoolep commented Oct 14, 2015

Choose a reason for hiding this comment

jwilder commented Oct 14, 2015

manishjain002 commented Nov 29, 2015

beckettsean commented Nov 30, 2015