Repository navigation
Feat: Build cortex-gateway and the Cortex ingestion endpoint #1758
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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,35 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
|
|
||
| "github.com/rossoctl/cortex/core/config" | ||
| "github.com/rossoctl/cortex/core/cost/pricing" | ||
| "github.com/rossoctl/cortex/core/sessionapi" | ||
| ) | ||
|
|
||
| // gatewayTokenEnv names the file that holds the bearer token cortex-gateway sends on | ||
| // POST /v1/ingest. The supervisor creates the file; Cortex only reads it. | ||
| const gatewayTokenEnv = "CORTEX_GATEWAY_TOKEN_FILE" | ||
|
|
||
| // ingestOptions turns on the ingestion endpoint in praxis mode. In direct mode it returns | ||
| // nil and reads neither the environment nor a file. | ||
| func ingestOptions(cfg *config.Config, rates pricing.Resolver, getenv func(string) string, readFile func(string) ([]byte, error)) ([]sessionapi.Option, error) { | ||
| if !cfg.PraxisEnabled() { | ||
| return nil, nil | ||
| } | ||
| path := getenv(gatewayTokenEnv) | ||
| if path == "" { | ||
| return nil, fmt.Errorf("praxis.enabled needs %s to name the gateway token file", gatewayTokenEnv) | ||
| } | ||
| b, err := readFile(path) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("read the gateway token file (%s): %w", gatewayTokenEnv, err) | ||
| } | ||
| token := strings.TrimSpace(string(b)) | ||
| if token == "" { | ||
| return nil, fmt.Errorf("the gateway token file %s (%s) is empty", path, gatewayTokenEnv) | ||
| } | ||
| return []sessionapi.Option{sessionapi.WithIngest(sessionapi.Ingest{Token: token, Rates: rates})}, nil | ||
| } |
This file contains hidden or 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,86 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "errors" | ||
| "net/http" | ||
| "net/http/httptest" | ||
| "strings" | ||
| "testing" | ||
|
|
||
| "github.com/rossoctl/cortex/core/config" | ||
| "github.com/rossoctl/cortex/core/session" | ||
| "github.com/rossoctl/cortex/core/sessionapi" | ||
| ) | ||
|
|
||
| func praxisCfg(enabled bool) *config.Config { | ||
| cfg := &config.Config{} | ||
| if enabled { | ||
| cfg.Praxis = &config.PraxisConfig{Enabled: true} | ||
| } | ||
| return cfg | ||
| } | ||
|
|
||
| func TestIngestOptions_DirectModeNone(t *testing.T) { | ||
| getenv := func(string) string { t.Fatal("getenv called in direct mode"); return "" } | ||
| readFile := func(string) ([]byte, error) { t.Fatal("readFile called in direct mode"); return nil, nil } | ||
| opts, err := ingestOptions(praxisCfg(false), nil, getenv, readFile) | ||
| if err != nil || opts != nil { | ||
| t.Fatalf("got %v, %v; want nil, nil", opts, err) | ||
| } | ||
| } | ||
|
|
||
| func TestIngestOptions_NeedsTokenFile(t *testing.T) { | ||
| _, err := ingestOptions(praxisCfg(true), nil, func(string) string { return "" }, | ||
| func(string) ([]byte, error) { return nil, nil }) | ||
| if err == nil || !strings.Contains(err.Error(), "CORTEX_GATEWAY_TOKEN_FILE") { | ||
| t.Fatalf("err = %v, want one that names CORTEX_GATEWAY_TOKEN_FILE", err) | ||
| } | ||
| _, err = ingestOptions(praxisCfg(true), nil, func(string) string { return "/x" }, | ||
| func(string) ([]byte, error) { return nil, errors.New("no such file") }) | ||
| if err == nil { | ||
| t.Fatal("an unreadable token file must be an error") | ||
| } | ||
| } | ||
|
|
||
| func TestIngestOptions_EmptyToken(t *testing.T) { | ||
| _, err := ingestOptions(praxisCfg(true), nil, func(string) string { return "/x" }, | ||
| func(string) ([]byte, error) { return []byte(" \n"), nil }) | ||
| if err == nil { | ||
| t.Fatal("an empty token must be an error") | ||
| } | ||
| } | ||
|
|
||
| func TestIngestOptions_ReadsToken(t *testing.T) { | ||
| opts, err := ingestOptions(praxisCfg(true), nil, | ||
| func(k string) string { | ||
| if k != "CORTEX_GATEWAY_TOKEN_FILE" { | ||
| t.Fatalf("getenv(%q)", k) | ||
| } | ||
| return "/tok" | ||
| }, | ||
| func(p string) ([]byte, error) { | ||
| if p != "/tok" { | ||
| t.Fatalf("readFile(%q)", p) | ||
| } | ||
| return []byte("abc\n"), nil | ||
| }) | ||
| if err != nil || len(opts) != 1 { | ||
| t.Fatalf("got %d options, %v; want 1, nil", len(opts), err) | ||
| } | ||
| srv := sessionapi.New(":0", session.New(0, 0, 0), opts...) | ||
| ts := httptest.NewServer(srv.Server().Handler) | ||
| defer ts.Close() | ||
| // The trailing newline is trimmed: "abc" is the token, "abc\n" from the file is not. | ||
| for token, want := range map[string]int{"abc": http.StatusNoContent, "abcd": http.StatusUnauthorized} { | ||
| req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/ingest", strings.NewReader("")) | ||
| req.Header.Set("Authorization", "Bearer "+token) | ||
| resp, err := http.DefaultClient.Do(req) | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| resp.Body.Close() | ||
| if resp.StatusCode != want { | ||
| t.Errorf("Bearer for %q: status %d, want %d", token, resp.StatusCode, want) | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or 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 hidden or 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 hidden or 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,65 @@ | ||
| package config | ||
|
|
||
| import ( | ||
| "strings" | ||
| "testing" | ||
|
|
||
| "gopkg.in/yaml.v3" | ||
| ) | ||
|
|
||
| func TestPraxis_DefaultOff(t *testing.T) { | ||
| var c Config | ||
| if err := yaml.Unmarshal([]byte("mode: proxy-sidecar\n"), &c); err != nil { | ||
| t.Fatalf("yaml: %v", err) | ||
| } | ||
| if c.PraxisEnabled() { | ||
| t.Fatal("PraxisEnabled() is true with no praxis block") | ||
| } | ||
| if err := yaml.Unmarshal([]byte("mode: proxy-sidecar\npraxis:\n enabled: true\n"), &c); err != nil { | ||
| t.Fatalf("yaml: %v", err) | ||
| } | ||
| if !c.PraxisEnabled() { | ||
| t.Fatal("praxis.enabled: true did not parse") | ||
| } | ||
| } | ||
|
|
||
| func TestPraxis_NeedsLoopbackOnly(t *testing.T) { | ||
| c := praxisConfig() | ||
| err := Validate(c) | ||
| if err == nil || !strings.Contains(err.Error(), "listener.bind_loopback_only") { | ||
| t.Fatalf("Validate = %v, want an error naming listener.bind_loopback_only", err) | ||
| } | ||
| c.Listener.BindLoopbackOnly = true | ||
| if err := Validate(c); err != nil { | ||
| t.Fatalf("Validate with bind_loopback_only: %v", err) | ||
| } | ||
| } | ||
|
|
||
| func TestPraxis_RefusedWithMTLS(t *testing.T) { | ||
| c := praxisConfig() | ||
| c.MTLS = &MTLSConfig{} | ||
| c.Listener.BindLoopbackOnly = true | ||
| err := Validate(c) | ||
| if err == nil || !strings.Contains(err.Error(), "mtls") { | ||
| t.Fatalf("Validate = %v, want an error naming mtls", err) | ||
| } | ||
| } | ||
|
|
||
| func TestPraxis_RefusedWithSessionsOff(t *testing.T) { | ||
| c := praxisConfig() | ||
| c.Listener.BindLoopbackOnly = true | ||
| off := false | ||
| c.Session.Enabled = &off | ||
| err := Validate(c) | ||
| if err == nil || !strings.Contains(err.Error(), "session.enabled") { | ||
| t.Fatalf("Validate = %v, want an error naming session.enabled", err) | ||
| } | ||
| } | ||
|
|
||
| // praxisConfig is a proxy-sidecar config that passes the listener checks, so | ||
| // only the praxis rules can refuse it. | ||
| func praxisConfig() *Config { | ||
| c := &Config{Mode: ModeProxySidecar, Praxis: &PraxisConfig{Enabled: true}} | ||
| c.Listener.ReverseProxyBackend = "127.0.0.1:8000" | ||
| return c | ||
| } |
This file contains hidden or 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.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
Use
http.NewRequestWithContextto fix thenoctxlint error.golangci-lint reports
net/http.NewRequest must not be calledon this line. Pass the test context so the request also stops when the test ends.Proposed fix
Committable suggestion
golangci-lint (2.15.0)
[error] 78-78: net/http.NewRequest must not be called. use net/http.NewRequestWithContext
(noctx)
Source: Linters/SAST tools