-
Notifications
You must be signed in to change notification settings - Fork 14
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* HTTP output * Rename output and stop in sigint and sigterm properly * Add tests * Various Makefile fixes * Add http server output to the readme * Avoid race condition in test
- Loading branch information
Showing
9 changed files
with
295 additions
and
24 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| // Licensed to Elasticsearch B.V. under one or more agreements. | ||
| // Elasticsearch B.V. licenses this file to you under the Apache 2.0 License. | ||
| // See the LICENSE file in the project root for more information. | ||
|
|
||
| package log | ||
|
|
||
| import ( | ||
| "go.uber.org/zap" | ||
| "go.uber.org/zap/zapcore" | ||
| ) | ||
|
|
||
| func NewLogger() (*zap.Logger, error) { | ||
| conf := zap.NewProductionConfig() | ||
| conf.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder | ||
| conf.Level = zap.NewAtomicLevelAt(zap.DebugLevel) | ||
| log, err := conf.Build() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return log, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,134 @@ | ||
| // Licensed to Elasticsearch B.V. under one or more agreements. | ||
| // Elasticsearch B.V. licenses this file to you under the Apache 2.0 License. | ||
| // See the LICENSE file in the project root for more information. | ||
| package httpserver | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "io/ioutil" | ||
| "net/http" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "go.uber.org/zap" | ||
|
|
||
| "github.com/elastic/stream/pkg/log" | ||
| "github.com/elastic/stream/pkg/output" | ||
| ) | ||
|
|
||
| func init() { | ||
| output.Register("http-server", New) | ||
| } | ||
|
|
||
| type Output struct { | ||
| logger *zap.SugaredLogger | ||
| opts *output.Options | ||
| server *http.Server | ||
| logChan chan []byte | ||
| ctx context.Context | ||
| } | ||
|
|
||
| func New(opts *output.Options) (output.Output, error) { | ||
| if opts.Addr == "" { | ||
| return nil, errors.New("a listen address is required") | ||
| } | ||
|
|
||
| if !(opts.HTTPServerOptions.TLSCertificate == "" && opts.HTTPServerOptions.TLSKey == "") && | ||
| !(opts.HTTPServerOptions.TLSCertificate != "" && opts.HTTPServerOptions.TLSKey != "") { | ||
| return nil, errors.New("both TLS certificate and key files must be defined") | ||
| } | ||
|
|
||
| if len(opts.HTTPServerOptions.ResponseHeaders)%2 != 0 { | ||
| return nil, errors.New("response headers must be a list of pairs") | ||
| } | ||
|
|
||
| logger, err := log.NewLogger() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| slogger := logger.Sugar().With("output", "http-server") | ||
|
|
||
| logChan := make(chan []byte) | ||
| server := &http.Server{ | ||
| Addr: opts.Addr, | ||
| ReadTimeout: opts.HTTPServerOptions.ReadTimeout, | ||
| WriteTimeout: opts.HTTPServerOptions.WriteTimeout, | ||
| MaxHeaderBytes: 1 << 20, | ||
| Handler: newHandler(opts, logChan, slogger), | ||
| } | ||
|
|
||
| return &Output{ | ||
| logger: slogger, | ||
| opts: opts, | ||
| server: server, | ||
| logChan: logChan, | ||
| }, nil | ||
| } | ||
|
|
||
| func (o *Output) DialContext(ctx context.Context) error { | ||
| o.ctx = ctx | ||
|
|
||
| if o.opts.TLSCertificate != "" && o.opts.TLSKey != "" { | ||
| go func() { o.logger.Info(o.server.ListenAndServeTLS(o.opts.TLSCertificate, o.opts.TLSKey)) }() | ||
| } else { | ||
| go func() { o.logger.Info(o.server.ListenAndServe()) }() | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (o *Output) Close() error { | ||
| defer close(o.logChan) | ||
|
|
||
| o.logger.Infow("shutting down http_server...") | ||
|
|
||
| ctx, cancel := context.WithTimeout(o.ctx, time.Second) | ||
| defer cancel() | ||
|
|
||
| return o.server.Shutdown(ctx) | ||
| } | ||
|
|
||
| func (o *Output) Write(b []byte) (int, error) { | ||
| if o.ctx == nil { | ||
| return 0, errors.New("DialContext needs to be called before Write can be used") | ||
| } | ||
|
|
||
| select { | ||
| case <-o.ctx.Done(): | ||
| o.logger.Infow("the output has been closed") | ||
| return 0, nil | ||
| case o.logChan <- b: | ||
| return len(b), nil | ||
| } | ||
| } | ||
|
|
||
| func newHandler(opts *output.Options, logChan <-chan []byte, logger *zap.SugaredLogger) http.HandlerFunc { | ||
| return func(w http.ResponseWriter, r *http.Request) { | ||
| b := <-logChan | ||
|
|
||
| defer r.Body.Close() | ||
| logger.Debug(strRequest(r)) | ||
|
|
||
| for i := 0; i < len(opts.HTTPServerOptions.ResponseHeaders); i += 2 { | ||
| w.Header().Add(opts.HTTPServerOptions.ResponseHeaders[i], opts.HTTPServerOptions.ResponseHeaders[i+1]) | ||
| } | ||
|
|
||
| _, _ = w.Write(b) | ||
| } | ||
| } | ||
|
|
||
| func strRequest(r *http.Request) string { | ||
| var b strings.Builder | ||
| b.WriteString("Request path: ") | ||
| b.WriteString(r.URL.String()) | ||
| b.WriteString(", Request Headers: ") | ||
| for k, v := range r.Header { | ||
| b.WriteString(fmt.Sprintf("'%s: %s' ", k, v)) | ||
| } | ||
| b.WriteString(", Request Body: ") | ||
| body, _ := ioutil.ReadAll(r.Body) | ||
| b.Write(body) | ||
| return b.String() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| // Licensed to Elasticsearch B.V. under one or more agreements. | ||
| // Elasticsearch B.V. licenses this file to you under the Apache 2.0 License. | ||
| // See the LICENSE file in the project root for more information. | ||
|
|
||
| package httpserver | ||
|
|
||
| import ( | ||
| "context" | ||
| "io/ioutil" | ||
| "net/http" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/elastic/stream/pkg/output" | ||
| ) | ||
|
|
||
| func TestHTTPServer(t *testing.T) { | ||
| cases := []struct { | ||
| description string | ||
| opts output.HTTPServerOptions | ||
| input []string | ||
| expectedOutput []string | ||
| expectedHeaders http.Header | ||
| }{ | ||
| { | ||
| description: "can get one log per response", | ||
| input: []string{"a", "b", "c"}, | ||
| expectedOutput: []string{"a", "b", "c"}, | ||
| }, | ||
| { | ||
| description: "returns expected response headers", | ||
| opts: output.HTTPServerOptions{ | ||
| ResponseHeaders: []string{"content-type", "custom"}, | ||
| }, | ||
| input: []string{"a"}, | ||
| expectedOutput: []string{"a"}, | ||
| expectedHeaders: http.Header{ | ||
| "Content-Type": []string{"custom"}, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| for _, tc := range cases { | ||
| tc := tc | ||
| t.Run(tc.description, func(t *testing.T) { | ||
| out, err := New(&output.Options{ | ||
| Addr: "127.0.0.1:1111", | ||
| HTTPServerOptions: tc.opts, | ||
| }) | ||
|
|
||
| require.NoError(t, err) | ||
| require.NoError(t, out.DialContext(context.Background())) | ||
|
|
||
| for i, in := range tc.input { | ||
| var n int | ||
| var werr error | ||
| var wg sync.WaitGroup | ||
| trigger := make(chan struct{}) | ||
| wg.Add(1) | ||
| go func(in string) { | ||
| defer wg.Done() | ||
|
|
||
| timeout := time.NewTimer(time.Second) | ||
| defer timeout.Stop() | ||
|
|
||
| select { | ||
| case <-timeout.C: | ||
| default: | ||
| close(trigger) | ||
| n, werr = out.Write([]byte(in)) | ||
| } | ||
| }(in) | ||
|
|
||
| <-trigger | ||
|
|
||
| resp, err := http.Get("http://127.0.0.1:1111") | ||
| require.NoError(t, err) | ||
| t.Cleanup(func() { resp.Body.Close() }) | ||
|
|
||
| wg.Wait() | ||
| require.NoError(t, werr) | ||
| assert.Equal(t, len(in), n) | ||
|
|
||
| body, err := ioutil.ReadAll(resp.Body) | ||
| require.NoError(t, err) | ||
|
|
||
| assert.Equal(t, tc.expectedOutput[i], string(body)) | ||
|
|
||
| for h, vs := range tc.expectedHeaders { | ||
| assert.EqualValues(t, vs, resp.Header[h]) | ||
| } | ||
| } | ||
|
|
||
| require.NoError(t, out.Close()) | ||
| }) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.