Feature/http sink #67
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package http | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "crypto/tls" | ||
| "encoding/base64" | ||
| "fmt" | ||
| sinkimpl "github.com/noctarius/timescaledb-event-streamer/internal/eventing/sink" | ||
| config "github.com/noctarius/timescaledb-event-streamer/spi/config" | ||
| "github.com/noctarius/timescaledb-event-streamer/spi/encoding" | ||
| "github.com/noctarius/timescaledb-event-streamer/spi/schema" | ||
| "github.com/noctarius/timescaledb-event-streamer/spi/sink" | ||
| "net/http" | ||
| "time" | ||
| ) | ||
|
|
||
| func init() { | ||
| sinkimpl.RegisterSink(config.Http, newHttpSink) | ||
| } | ||
|
|
||
| func basicAuth( | ||
| username, password string, | ||
| ) string { | ||
| auth := username + ":" + password | ||
| return base64.StdEncoding.EncodeToString([]byte(auth)) | ||
| } | ||
|
|
||
| type httpSink struct { | ||
| client *http.Client | ||
| encoder *encoding.JsonEncoder | ||
| address *string | ||
| headers *http.Header | ||
| } | ||
|
|
||
| func newHttpSink( | ||
| c *config.Config, | ||
| ) (sink.Sink, error) { | ||
|
|
||
| transport := http.DefaultTransport.(*http.Transport).Clone() | ||
| if config.GetOrDefault(c, config.PropertyHttpTlsEnabled, false) { | ||
| transport.TLSClientConfig = &tls.Config{ | ||
| InsecureSkipVerify: config.GetOrDefault( | ||
| c, config.PropertyHttpTlsSkipVerify, false, | ||
| ), | ||
| ClientAuth: config.GetOrDefault( | ||
| c, config.PropertyHttpTlsClientAuth, tls.NoClientCert, | ||
| ), | ||
| } | ||
| } | ||
|
|
||
| address := config.GetOrDefault(c, config.PropertyHttpUrl, "http://localhost:80") | ||
| headers := make(http.Header) | ||
|
|
||
| authenticationType := config.GetOrDefault(c, config.PropertyHttpAuthenticationType, "none") | ||
| switch config.HttpAuthenticationType(authenticationType) { | ||
| case config.BasicAuthentication: | ||
| { | ||
| headers.Add("Authorization", | ||
| fmt.Sprintf("Basic %s", | ||
| basicAuth(config.GetOrDefault(c, config.PropertyHttpBasicAuthenticationUsername, ""), | ||
| config.GetOrDefault(c, config.PropertyHttpBasicAuthenticationPassword, ""), | ||
| ), | ||
| ), | ||
| ) | ||
| } | ||
| case config.HeaderAuthentication: | ||
| { | ||
| headers.Add(config.GetOrDefault(c, config.PropertyHttpHeaderAuthenticationHeaderName, ""), | ||
| config.GetOrDefault(c, config.PropertyHttpHeaderAuthenticationHeaderValue, ""), | ||
| ) | ||
| } | ||
| case config.NoneAuthentication: | ||
| { | ||
| } | ||
| default: | ||
| { | ||
| return nil, fmt.Errorf("http AuthenticationType '%s' doesn't exist", authenticationType) | ||
| } | ||
| } | ||
|
|
||
| return &httpSink{ | ||
| client: &http.Client{Transport: transport}, | ||
| encoder: encoding.NewJsonEncoderWithConfig(c), | ||
| address: &address, | ||
| headers: &headers, | ||
| }, nil | ||
| } | ||
|
|
||
| func (h *httpSink) Start() error { | ||
| return nil | ||
| } | ||
|
|
||
| func (h *httpSink) Stop() error { | ||
| h.client.CloseIdleConnections() | ||
| return nil | ||
| } | ||
|
|
||
| func (h *httpSink) Emit( | ||
| _ sink.Context, _ time.Time, topicName string, key, envelope schema.Struct, | ||
| ) error { | ||
| delete(envelope, "schema") | ||
|
There was a problem hiding this comment. I wonder if there should be a general property to disable schema sending. I think right now it'd be the only sink not including it. WDYT? There was a problem hiding this comment. I think this mainly depends on where you'd like to take the project. Would you like to have everything or almost everything as a configurable parameter? I can also see a world where the package always emits everything and it's up to the receiver to discard whatever is not needed or if one would like to avoid unnecessary traffic then one could transform the payload with a plugin before emitting. I think you probably have better things to work on in this project then having to make I have to be honest and tell you that I've started the http sink by copying the For now I think it'd be best if all the sinks behaved the same way (aside from maybe the I've pushed a commit that adds back in the schema for this sink as well. I'll use a plugin to transform the payload before sending to the sink anyway, so I'll just omit it there. There was a problem hiding this comment. Yeah I agree. I think it could be interesting since the schema creates quite some overhead and traffic. I also think it may be better to provide the option to create a separate stream of schema data, which could be routed to a schema registry, but I never used one before so I wouldn't have enough insight into it. I think for now we can just keep it in and make it up to the receiver to discard it. For traffic reasons it may be interesting to eventually provide the option. Tbh, even stdout should support it, since the idea (next to testing) was to pipe the output somewhere else. The reason why you can send any log message to stderr 😁 There was a problem hiding this comment.
Yeah, this is probably the biggest argument for making it optional.
What are your thoughts on discarding it with a Plugin before sending it? We're streaming anywhere between 10k - 100k per second with |
||
| payload, err := h.encoder.Marshal(envelope) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| req, err := http.NewRequest("POST", *h.address, bytes.NewBuffer(payload)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| req.Header = *h.headers | ||
| _, err = h.client.Do(req) | ||
| return err | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If you have a separate transport, wouldn't it make sense to remove the protocol part here? Alternatively instead of using the TLS enabled prop, you could just check the protocol and if it's HTTPS go and assume TLS enabled (and just read the remaining properties).
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You're absolutely right, I was just hacking it together so the hard-coded one stayed in. I've pushed the commit where we infer the usage of TLS by the prefix of the url and if it's present then take the TLS settings into account.