Skip to content

Commit c767a4d

Browse files
xrstfxrstf
committed
turn cluster resolver middleware into a mux that can recognize cluster names in path mappings
On-behalf-of: @SAP [email protected]
1 parent 8181da0 commit c767a4d

File tree

8 files changed

+205
-150
lines changed

8 files changed

+205
-150
lines changed

‎cmd/sharded-test-server/frontproxy.go‎

Lines changed: 3 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ import (
3939
"k8s.io/klog/v2"
4040

4141
"github.com/kcp-dev/kcp/cmd/test-server/helpers"
42+
"github.com/kcp-dev/kcp/pkg/server/proxy/types"
4243
kcpclientset "github.com/kcp-dev/kcp/sdk/client/clientset/versioned/cluster"
4344
kcptestingserver "github.com/kcp-dev/kcp/sdk/testing/server"
4445
"github.com/kcp-dev/kcp/sdk/testing/third_party/library-go/crypto"
@@ -66,15 +67,7 @@ func startFrontProxy(
6667

6768
logger := klog.FromContext(ctx)
6869

69-
type mappingEntry struct {
70-
Path string `json:"path"`
71-
Backend string `json:"backend"`
72-
BackendServerCA string `json:"backend_server_ca"`
73-
ProxyClientCert string `json:"proxy_client_cert"`
74-
ProxyClientKey string `json:"proxy_client_key"`
75-
}
76-
77-
mappings := []mappingEntry{
70+
mappings := []types.PathMapping{
7871
{
7972
Path: "/services/",
8073
// TODO: support multiple virtual workspace backend servers
@@ -85,7 +78,7 @@ func startFrontProxy(
8578
},
8679
{
8780
Path: "/clusters/",
88-
// TODO: support multiple shard backend servers
81+
// this path is not actually used, since shard URLs are determined based on the Shard
8982
Backend: "https://localhost:6444",
9083
BackendServerCA: filepath.Join(workDirPath, ".kcp", "serving-ca.crt"),
9184
ProxyClientCert: filepath.Join(workDirPath, ".kcp-front-proxy", "requestheader.crt"),

‎pkg/proxy/lookup/lookup.go‎

Lines changed: 109 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -30,63 +30,138 @@ import (
3030
"github.com/kcp-dev/logicalcluster/v3"
3131

3232
kcpauthorization "github.com/kcp-dev/kcp/pkg/authorization"
33+
"github.com/kcp-dev/kcp/pkg/index"
3334
proxyindex "github.com/kcp-dev/kcp/pkg/proxy/index"
35+
"github.com/kcp-dev/kcp/pkg/server/proxy/types"
3436
)
3537

36-
func WithClusterResolver(delegate http.Handler, index proxyindex.Index) http.Handler {
37-
return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
38-
var cs = strings.SplitN(strings.TrimLeft(req.URL.Path, "/"), "/", 3)
39-
if len(cs) < 2 || cs[0] != "clusters" {
40-
delegate.ServeHTTP(w, req)
41-
return
38+
func WithClusterResolver(delegate http.Handler, mappings []types.PathMapping, index proxyindex.Index) http.Handler {
39+
mux := http.NewServeMux()
40+
41+
// fallback for all unrecognized URLs
42+
mux.Handle("/", delegate)
43+
44+
// Use the extra path mappings as an additional source of cluster names in URLs;
45+
// it's okay for a virtual workspace URL to not match here or to not have a
46+
// cluster placeholder in its URL pattern, since the default handler will simply
47+
// forward the request unchanged (and most likely, unauthenticated).
48+
49+
// We can use the same handler for all mappings, since the actual muxing to
50+
// the destinations happens later in proxy.HttpHandler; here we only care about
51+
// detecting the cluster name.
52+
mappingHandler := newMappingHandler(delegate, index)
53+
54+
for _, mapping := range mappings {
55+
p := strings.TrimRight(mapping.Path, "/")
56+
57+
// Even though we know how to handle the "special" core clusters path,
58+
// the mapping provides additional PKI configuration that is not available
59+
// by just looking up the cluster in the index and figuring out the
60+
// target shard. That's why it's required to configure /clusters/ in the
61+
// front-proxy mappings and since admins could choose not to include it,
62+
// we only enable the built-in clusterResolveHandler if we actually find
63+
// an appropriate mapping.
64+
if p == "/clusters" {
65+
// we know how to parse cluster URLs
66+
resolveHandler := newClusterResolveHandler(delegate, index)
67+
mux.HandleFunc("/clusters/{cluster}", resolveHandler)
68+
mux.HandleFunc("/clusters/{cluster}/{trail...}", resolveHandler)
69+
} else {
70+
// mappings are configured with *prefixes*; in order to match both exact matches
71+
// and prefix matches (i.e. if "/foo" is configured, both "/foo" and "/foo/bar"
72+
// must match), each mapping is added twice to the mux.
73+
mux.HandleFunc(p, mappingHandler)
74+
mux.HandleFunc(p+"/{trail...}", mappingHandler)
4275
}
76+
}
4377

44-
ctx := req.Context()
45-
logger := klog.FromContext(ctx)
46-
attributes, err := filters.GetAuthorizerAttributes(ctx)
47-
if err != nil {
48-
responsewriters.InternalError(w, req, err)
49-
return
50-
}
78+
return mux
79+
}
5180

52-
clusterPath := logicalcluster.NewPath(cs[1])
53-
if !clusterPath.IsValid() {
54-
// this includes wildcards
55-
logger.WithValues("requestPath", req.URL.Path).V(4).Info("Invalid cluster path")
56-
responsewriters.Forbidden(req.Context(), attributes, w, req, kcpauthorization.WorkspaceAccessNotPermittedReason, kubernetesscheme.Codecs)
57-
return
58-
}
81+
func newClusterResolveHandler(delegate http.Handler, index proxyindex.Index) http.HandlerFunc {
82+
return func(w http.ResponseWriter, req *http.Request) {
83+
clusterName := req.PathValue("cluster")
5984

60-
result, found := index.LookupURL(clusterPath)
61-
if result.ErrorCode != 0 {
62-
http.Error(w, "Not available.", result.ErrorCode)
63-
return
64-
}
65-
if !found {
66-
logger.WithValues("clusterPath", clusterPath).V(4).Info("Unknown cluster path")
67-
responsewriters.Forbidden(req.Context(), attributes, w, req, kcpauthorization.WorkspaceAccessNotPermittedReason, kubernetesscheme.Codecs)
85+
req, result := resolveClusterName(w, req, index, clusterName)
86+
if req == nil {
6887
return
6988
}
89+
7090
shardURL, err := url.Parse(result.URL)
7191
if err != nil {
7292
responsewriters.InternalError(w, req, err)
7393
return
7494
}
7595

76-
logger.WithValues("from", "/clusters/"+cs[1], "to", shardURL).V(4).Info("Redirecting")
96+
ctx := req.Context()
97+
98+
logger := klog.FromContext(ctx)
99+
logger.WithValues("from", "/clusters/"+clusterName, "to", shardURL).V(4).Info("Redirecting")
77100

101+
shardURL.RawQuery = req.URL.RawQuery
78102
shardURL.Path = strings.TrimSuffix(shardURL.Path, "/")
79-
if len(cs) == 3 {
80-
shardURL.Path += "/" + cs[2]
103+
if trail := req.PathValue("trail"); len(trail) != 0 {
104+
shardURL.Path += "/" + trail
81105
}
82106

83107
ctx = WithShardURL(ctx, shardURL)
84-
ctx = WithClusterName(ctx, result.Cluster)
85-
ctx = WithWorkspaceType(ctx, result.Type)
86108
req = req.WithContext(ctx)
87109

88110
delegate.ServeHTTP(w, req)
89-
})
111+
}
112+
}
113+
114+
func newMappingHandler(delegate http.Handler, index proxyindex.Index) http.HandlerFunc {
115+
return func(w http.ResponseWriter, req *http.Request) {
116+
// not every virtual workspace and/or every mapping has a {cluster} in its URL;
117+
// also wildcard requests have to be passed without lookup
118+
clusterName := req.PathValue("cluster")
119+
if clusterName == "" || clusterName == "*" {
120+
delegate.ServeHTTP(w, req)
121+
return
122+
}
123+
124+
req, _ = resolveClusterName(w, req, index, clusterName)
125+
if req == nil {
126+
return
127+
}
128+
129+
delegate.ServeHTTP(w, req)
130+
}
131+
}
132+
133+
func resolveClusterName(w http.ResponseWriter, req *http.Request, index proxyindex.Index, clusterName string) (*http.Request, *index.Result) {
134+
ctx := req.Context()
135+
logger := klog.FromContext(ctx)
136+
attributes, err := filters.GetAuthorizerAttributes(ctx)
137+
if err != nil {
138+
responsewriters.InternalError(w, req, err)
139+
return nil, nil
140+
}
141+
142+
clusterPath := logicalcluster.NewPath(clusterName)
143+
if !clusterPath.IsValid() {
144+
// this includes wildcards
145+
logger.WithValues("requestPath", req.URL.Path).V(4).Info("Invalid cluster path")
146+
responsewriters.Forbidden(req.Context(), attributes, w, req, kcpauthorization.WorkspaceAccessNotPermittedReason, kubernetesscheme.Codecs)
147+
return nil, nil
148+
}
149+
150+
result, found := index.LookupURL(clusterPath)
151+
if result.ErrorCode != 0 {
152+
http.Error(w, "Not available.", result.ErrorCode)
153+
return nil, nil
154+
}
155+
if !found {
156+
logger.WithValues("clusterPath", clusterPath).V(4).Info("Unknown cluster path")
157+
responsewriters.Forbidden(req.Context(), attributes, w, req, kcpauthorization.WorkspaceAccessNotPermittedReason, kubernetesscheme.Codecs)
158+
return nil, nil
159+
}
160+
161+
ctx = WithClusterName(ctx, result.Cluster)
162+
ctx = WithWorkspaceType(ctx, result.Type)
163+
164+
return req.WithContext(ctx), &result
90165
}
91166

92167
type lookupKey int

‎pkg/proxy/mapping.go‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -29,32 +29,31 @@ import (
2929
"k8s.io/component-base/metrics/legacyregistry"
3030
"k8s.io/klog/v2"
3131

32-
"github.com/kcp-dev/kcp/pkg/proxy/index"
3332
"github.com/kcp-dev/kcp/pkg/server/proxy"
33+
"github.com/kcp-dev/kcp/pkg/server/proxy/types"
3434
)
3535

36-
func loadMappings(filename string) ([]proxy.PathMapping, error) {
36+
func loadMappings(filename string) ([]types.PathMapping, error) {
3737
mappingData, err := os.ReadFile(filename)
3838
if err != nil {
3939
return nil, err
4040
}
4141

42-
var mapping []proxy.PathMapping
42+
var mapping []types.PathMapping
4343
if err := yaml.Unmarshal(mappingData, &mapping); err != nil {
4444
return nil, err
4545
}
4646

4747
return mapping, nil
4848
}
4949

50-
func isShardMapping(m proxy.PathMapping) bool {
50+
func isShardMapping(m types.PathMapping) bool {
5151
return m.Path == "/clusters/"
5252
}
5353

54-
func NewHandler(ctx context.Context, mappings []proxy.PathMapping, index index.Index) (http.Handler, error) {
54+
func NewHandler(ctx context.Context, mappings []types.PathMapping) (http.Handler, error) {
5555
handlers := proxy.HttpHandler{
56-
Index: index,
57-
Mappings: proxy.HttpHandlerMappings{
56+
Mappings: types.HttpHandlerMappings{
5857
{
5958
Weight: 0,
6059
Path: "/metrics",
@@ -108,7 +107,7 @@ func NewHandler(ctx context.Context, mappings []proxy.PathMapping, index index.I
108107
if m.Path == "/" {
109108
handlers.DefaultHandler = handler
110109
} else {
111-
handlers.Mappings = append(handlers.Mappings, proxy.HttpHandlerMapping{
110+
handlers.Mappings = append(handlers.Mappings, types.HttpHandlerMapping{
112111
Weight: len(m.Path),
113112
Path: m.Path,
114113
Handler: handler,

‎pkg/proxy/server.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@ func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) {
9393
// interface.
9494
s.IndexController = index.NewController(ctx, s.KcpSharedInformerFactory.Core().V1alpha1().Shards(), getClientFunc)
9595

96-
handler, err := NewHandler(ctx, mappings, s.IndexController)
96+
handler, err := NewHandler(ctx, mappings)
9797
if err != nil {
9898
return s, err
9999
}
@@ -131,7 +131,7 @@ func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) {
131131

132132
if hasShardMapping {
133133
// This middleware must happen before the authentication.
134-
handler = lookup.WithClusterResolver(handler, s.IndexController)
134+
handler = lookup.WithClusterResolver(handler, mappings, s.IndexController)
135135
}
136136

137137
requestInfoFactory := requestinfo.NewFactory()

‎pkg/server/localproxy.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ import (
4242
"github.com/kcp-dev/kcp/pkg/proxy/lookup"
4343
"github.com/kcp-dev/kcp/pkg/server/filters"
4444
"github.com/kcp-dev/kcp/pkg/server/proxy"
45+
"github.com/kcp-dev/kcp/pkg/server/proxy/types"
4546
corev1alpha1 "github.com/kcp-dev/kcp/sdk/apis/core/v1alpha1"
4647
tenancyv1alpha1 "github.com/kcp-dev/kcp/sdk/apis/tenancy/v1alpha1"
4748
corev1alpha1informers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/core/v1alpha1"
@@ -211,7 +212,7 @@ func WithLocalProxy(
211212
}
212213

213214
// If additional mappings file is provided, read it and add the mappings to the handler
214-
handlers, err := NewLocalProxyHandler(defaultHandlerFunc, indexState, additionalMappingsFile)
215+
handlers, err := NewLocalProxyHandler(defaultHandlerFunc, additionalMappingsFile)
215216
if err != nil {
216217
return nil, fmt.Errorf("failed to create local proxy handler: %w", err)
217218
}
@@ -223,8 +224,8 @@ func WithLocalProxy(
223224
// This function is very similar to proxy/mapping.go.NewHandler.
224225
// If we want to re-use that code, we basically would be merging proxy with server packages.
225226
// Which is not desirable at the point of writing (2024-10-26), but might be in the future.
226-
func NewLocalProxyHandler(defaultHandler http.Handler, index index.Index, additionalMappingsFile string) (http.Handler, error) {
227-
mapping := []proxy.PathMapping{}
227+
func NewLocalProxyHandler(defaultHandler http.Handler, additionalMappingsFile string) (http.Handler, error) {
228+
mapping := []types.PathMapping{}
228229
if additionalMappingsFile != "" {
229230
mappingData, err := os.ReadFile(additionalMappingsFile)
230231
if err != nil {
@@ -237,8 +238,7 @@ func NewLocalProxyHandler(defaultHandler http.Handler, index index.Index, additi
237238
}
238239

239240
handlers := proxy.HttpHandler{
240-
Index: index,
241-
Mappings: proxy.HttpHandlerMappings{
241+
Mappings: types.HttpHandlerMappings{
242242
{
243243
Weight: 0,
244244
Path: "/metrics",
@@ -280,7 +280,7 @@ func NewLocalProxyHandler(defaultHandler http.Handler, index index.Index, additi
280280

281281
handler = withProxyAuthHeaders(handler, userHeader, groupHeader, extraHeaderPrefix)
282282

283-
handlers.Mappings = append(handlers.Mappings, proxy.HttpHandlerMapping{
283+
handlers.Mappings = append(handlers.Mappings, types.HttpHandlerMapping{
284284
Weight: len(m.Path),
285285
Path: m.Path,
286286
Handler: handler,

0 commit comments

Comments
 (0)