-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add support for exporting logs via wasitellog.Exporter (#51)
Signed-off-by: Joonas Bergius <[email protected]>
- Loading branch information
Showing
11 changed files
with
881 additions
and
15 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| package wasitellog | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "net/http" | ||
| "net/url" | ||
|
|
||
| "go.wasmcloud.dev/component/net/wasihttp" | ||
| "go.wasmcloud.dev/component/x/wasitel/wasitellog/internal/types" | ||
| ) | ||
|
|
||
| type client struct { | ||
| config config | ||
| httpClient *http.Client | ||
| } | ||
|
|
||
| func newClient(opts ...Option) *client { | ||
| cfg := newConfig(opts...) | ||
|
|
||
| wasiTransport := &wasihttp.Transport{} | ||
| httpClient := &http.Client{Transport: wasiTransport} | ||
|
|
||
| return &client{ | ||
| config: cfg, | ||
| httpClient: httpClient, | ||
| } | ||
| } | ||
|
|
||
| func (c *client) UploadLogs(ctx context.Context, logs []*types.ResourceLogs) error { | ||
| if len(logs) == 0 { | ||
| return nil | ||
| } | ||
|
|
||
| export := &types.ExportLogsServiceRequest{ | ||
| ResourceLogs: logs, | ||
| } | ||
|
|
||
| body, err := json.Marshal(export) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to serialize export request to JSON: %w", err) | ||
| } | ||
|
|
||
| u := c.getUrl() | ||
| req, err := http.NewRequest(http.MethodPost, u.String(), bytes.NewBuffer(body)) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create request to %q: %w", u.String(), err) | ||
| } | ||
| req.Header.Set("Content-Type", "application/json") | ||
|
|
||
| _, err = c.httpClient.Do(req) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to request %q: %w", u.String(), err) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (c *client) getUrl() url.URL { | ||
| scheme := "http" | ||
| if !c.config.Insecure { | ||
| scheme = "https" | ||
| } | ||
| u := url.URL{ | ||
| Scheme: scheme, | ||
| Host: c.config.Endpoint, | ||
| Path: c.config.Path, | ||
| } | ||
| return u | ||
| } |
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,74 @@ | ||
| package wasitellog | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "net/url" | ||
| ) | ||
|
|
||
| const ( | ||
| // DefaultPort is the default HTTP port of the collector. | ||
| DefaultPort uint16 = 4318 | ||
| // DefaultHost is the host address the client will attempt | ||
| // connect to if no collector address is provided. | ||
| DefaultHost string = "localhost" | ||
| // DefaultPath is a default URL path for endpoint that receives logs | ||
| DefaultPath string = "/v1/logs" | ||
| ) | ||
|
|
||
| type config struct { | ||
| Endpoint string | ||
| Insecure bool | ||
| Path string | ||
| } | ||
|
|
||
| func newConfig(opts ...Option) config { | ||
| cfg := config{ | ||
| Insecure: true, | ||
| Endpoint: fmt.Sprintf("%s:%d", DefaultHost, DefaultPort), | ||
| Path: DefaultPath, | ||
| } | ||
| for _, opt := range opts { | ||
| cfg = opt.apply(cfg) | ||
| } | ||
| return cfg | ||
| } | ||
|
|
||
| type Option interface { | ||
| apply(config) config | ||
| } | ||
|
|
||
| func newWrappedOption(fn func(config) config) Option { | ||
| return &wrappedOption{fn: fn} | ||
| } | ||
|
|
||
| type wrappedOption struct { | ||
| fn func(config) config | ||
| } | ||
|
|
||
| func (o *wrappedOption) apply(cfg config) config { | ||
| return o.fn(cfg) | ||
| } | ||
|
|
||
| func WithEndpoint(endpoint string) Option { | ||
| return newWrappedOption(func(cfg config) config { | ||
| cfg.Endpoint = endpoint | ||
| return cfg | ||
| }) | ||
| } | ||
|
|
||
| func WithEndpointURL(eu string) Option { | ||
| return newWrappedOption(func(cfg config) config { | ||
| u, err := url.Parse(eu) | ||
| if err != nil { | ||
| return cfg | ||
| } | ||
|
|
||
| cfg.Endpoint = u.Host | ||
| cfg.Path = u.Path | ||
| if u.Scheme != "https" { | ||
| cfg.Insecure = true | ||
| } | ||
|
|
||
| return cfg | ||
| }) | ||
| } |
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,64 @@ | ||
| package wasitellog | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "sync" | ||
|
|
||
| logsdk "go.opentelemetry.io/otel/sdk/log" | ||
| "go.wasmcloud.dev/component/x/wasitel/wasitellog/internal/convert" | ||
| ) | ||
|
|
||
| func New(opts ...Option) (*Exporter, error) { | ||
| client := newClient(opts...) | ||
| return &Exporter{ | ||
| client: client, | ||
| }, nil | ||
| } | ||
|
|
||
| var _ logsdk.Exporter = (*Exporter)(nil) | ||
|
|
||
| type Exporter struct { | ||
| client *client | ||
| stopped bool | ||
| stoppedMu sync.RWMutex | ||
| } | ||
|
|
||
| func (e *Exporter) Export(ctx context.Context, logs []logsdk.Record) error { | ||
| err := ctx.Err() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // Check whether the exporter has been told to Shutdown | ||
| e.stoppedMu.RLock() | ||
| stopped := e.stopped | ||
| e.stoppedMu.RUnlock() | ||
| if stopped { | ||
| return nil | ||
| } | ||
|
|
||
| // Check whether there's anything to export | ||
| converted := convert.ResourceLogs(logs) | ||
| if len(converted) == 0 { | ||
| return nil | ||
| } | ||
|
|
||
| err = e.client.UploadLogs(ctx, converted) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to export spans: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (e *Exporter) Shutdown(ctx context.Context) error { | ||
| e.stoppedMu.Lock() | ||
| e.stopped = true | ||
| e.stoppedMu.Unlock() | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (e *Exporter) ForceFlush(ctx context.Context) error { | ||
| return nil | ||
| } |
Oops, something went wrong.