mirror of
https://github.com/zitadel/zitadel.git
synced 2025-08-12 10:37:32 +00:00
enable request headers
This commit is contained in:
@@ -28,6 +28,12 @@ Telemetry:
|
||||
# If you change this configuration at runtime, remaining data that is not successfully delivered to the old endpoints is sent to the new endpoints.
|
||||
Endpoints:
|
||||
- https://httpbin.org/post
|
||||
# These headers are sent with every request to the configured endpoints.
|
||||
Headers:
|
||||
# single-value: "single-value"
|
||||
# multi-value:
|
||||
# - "multi-value-1"
|
||||
# - "multi-value-2"
|
||||
|
||||
# Port ZITADEL will listen on
|
||||
Port: 8080
|
||||
|
@@ -8,6 +8,11 @@ Telemetry:
|
||||
Enabled: true
|
||||
Endpoints:
|
||||
- http://localhost:8081
|
||||
Headers:
|
||||
single-value: "single-value"
|
||||
multi-value:
|
||||
- "multi-value-1"
|
||||
- "multi-value-2"
|
||||
|
||||
FirstInstance:
|
||||
Org:
|
||||
|
@@ -21,10 +21,8 @@ func InitChannel(ctx context.Context, cfg Config) (channels.NotificationChannel,
|
||||
|
||||
logging.Debug("successfully initialized webhook json channel")
|
||||
return channels.HandleMessageFunc(func(message channels.Message) error {
|
||||
|
||||
requestCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
msg, ok := message.(*messages.JSON)
|
||||
if !ok {
|
||||
return errors.ThrowInternal(nil, "WEBH-K686U", "message is not JSON")
|
||||
@@ -33,27 +31,24 @@ func InitChannel(ctx context.Context, cfg Config) (channels.NotificationChannel,
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(requestCtx, cfg.Method, cfg.CallURL, strings.NewReader(payload))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if cfg.Headers != nil {
|
||||
req.Header = cfg.Headers
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = resp.Body.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return errors.ThrowUnknown(fmt.Errorf("calling url %s returned %s", cfg.CallURL, resp.Status), "WEBH-LBxU0", "webhook didn't return a success status")
|
||||
}
|
||||
|
||||
logging.WithFields("calling_url", cfg.CallURL, "method", cfg.Method).Debug("webhook called")
|
||||
return nil
|
||||
}), nil
|
||||
|
@@ -1,12 +1,14 @@
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/url"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
CallURL string
|
||||
Method string
|
||||
Headers http.Header
|
||||
}
|
||||
|
||||
func (w *Config) Validate() error {
|
||||
|
@@ -30,15 +30,16 @@ const (
|
||||
type TelemetryPusherConfig struct {
|
||||
Enabled bool
|
||||
Endpoints []string
|
||||
Headers http.Header
|
||||
}
|
||||
|
||||
type telemetryPusher struct {
|
||||
crdb.StatementHandler
|
||||
cfg TelemetryPusherConfig
|
||||
commands *command.Commands
|
||||
queries *NotificationQueries
|
||||
metricSuccessfulDeliveriesJSON string
|
||||
metricFailedDeliveriesJSON string
|
||||
endpoints []string
|
||||
}
|
||||
|
||||
func NewTelemetryPusher(
|
||||
@@ -56,7 +57,7 @@ func NewTelemetryPusher(
|
||||
if telemetryCfg.Enabled {
|
||||
handlerCfg.Reducers = p.reducers()
|
||||
}
|
||||
p.endpoints = telemetryCfg.Endpoints
|
||||
p.cfg = telemetryCfg
|
||||
p.StatementHandler = crdb.NewStatementHandler(ctx, handlerCfg)
|
||||
p.commands = commands
|
||||
p.queries = queries
|
||||
@@ -129,12 +130,13 @@ func (t *telemetryPusher) pushMilestone(ctx context.Context, event *pseudo.Sched
|
||||
if alreadyHandled {
|
||||
return nil
|
||||
}
|
||||
for _, endpoint := range t.endpoints {
|
||||
for _, endpoint := range t.cfg.Endpoints {
|
||||
if err := types.SendJSON(
|
||||
ctx,
|
||||
webhook.Config{
|
||||
CallURL: endpoint,
|
||||
Method: http.MethodPost,
|
||||
Headers: t.cfg.Headers,
|
||||
},
|
||||
t.queries.GetFileSystemProvider,
|
||||
t.queries.GetLogProvider,
|
||||
@@ -146,5 +148,5 @@ func (t *telemetryPusher) pushMilestone(ctx context.Context, event *pseudo.Sched
|
||||
return err
|
||||
}
|
||||
}
|
||||
return t.commands.MilestonePushed(ctx, ms.Type, t.endpoints, ms.PrimaryDomain)
|
||||
return t.commands.MilestonePushed(ctx, ms.Type, t.cfg.Endpoints, ms.PrimaryDomain)
|
||||
}
|
||||
|
@@ -9,6 +9,7 @@ import (
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -25,6 +26,12 @@ func TestServer_TelemetryPushMilestones(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
if r.Header.Get("single-value") != "single-value" {
|
||||
t.Error("single-value header not set")
|
||||
}
|
||||
if reflect.DeepEqual(r.Header.Get("multi-value"), "multi-value-1,multi-value-2") {
|
||||
t.Error("single-value header not set")
|
||||
}
|
||||
bodies <- body
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
|
Reference in New Issue
Block a user