Skip to content

Commit 50857d5

Browse files
authored
Update dubbo gateway agent code logic (#861)
1 parent 42b11f9 commit 50857d5

File tree

20 files changed

+276
-473
lines changed

20 files changed

+276
-473
lines changed

dubbod/discovery/cmd/dubbo-agent/app/cmd.go

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,8 @@ import (
2020
"context"
2121
"errors"
2222
"fmt"
23-
"net/netip"
24-
2523
"github.com/apache/dubbo-kubernetes/pkg/log"
24+
"net/netip"
2625

2726
"github.com/apache/dubbo-kubernetes/dubbod/discovery/cmd/dubbo-agent/options"
2827
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/util/network"
@@ -65,7 +64,7 @@ func NewRootCommand(sds dubboagent.SDSServiceFactory) *cobra.Command {
6564
func newProxyCommand(sds dubboagent.SDSServiceFactory) *cobra.Command {
6665
return &cobra.Command{
6766
Use: "proxy",
68-
Short: "xDS proxy agent",
67+
Short: "XDS proxy agent",
6968
FParseErrWhitelist: cobra.FParseErrWhitelist{
7069
UnknownFlags: true,
7170
},
@@ -99,6 +98,7 @@ func newProxyCommand(sds dubboagent.SDSServiceFactory) *cobra.Command {
9998

10099
agentOptions := options.NewAgentOptions(&proxyArgs, proxyConfig, sds)
101100
agent := dubboagent.NewAgent(proxyConfig, agentOptions, secOpts)
101+
102102
ctx, cancel := context.WithCancelCause(context.Background())
103103
defer cancel(errors.New("application shutdown"))
104104
defer agent.Close()
@@ -110,9 +110,7 @@ func newProxyCommand(sds dubboagent.SDSServiceFactory) *cobra.Command {
110110
if err != nil {
111111
return err
112112
}
113-
114113
wait()
115-
116114
return nil
117115
},
118116
}

dubbod/discovery/docker/dockerfile.proxy

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,6 @@
1515

1616
FROM gcr.io/distroless/static:debug
1717
COPY bin/dubbo-agent /usr/local/bin/dubbo-agent
18-
COPY bin/pixiugateway /usr/local/bin/pixiugateway
18+
COPY bin/pixiu-gateway /usr/local/bin/pixiu-gateway
1919
USER 9999:9999
2020
ENTRYPOINT ["/usr/local/bin/dubbo-agent"]

dubbod/discovery/pkg/config/kube/crdclient/client.go

Lines changed: 24 additions & 130 deletions
Original file line numberDiff line numberDiff line change
@@ -156,7 +156,7 @@ func (cl *Client) addCRD(name string, opts krt.OptionsBuilder) {
156156
cl.logger.Debugf("addCRD: adding CRD %q", name)
157157
s, f := cl.schemasByCRDName[name]
158158
if !f {
159-
cl.logger.Debugf("Added resource that we are not watching: %v", name)
159+
cl.logger.Debugf("added resource that we are not watching: %v", name)
160160
return
161161
}
162162
resourceGVK := s.GroupVersionKind()
@@ -201,23 +201,18 @@ func (cl *Client) addCRD(name string, opts krt.OptionsBuilder) {
201201
ObjectTransform: transform,
202202
FieldSelector: fieldSelector,
203203
}
204-
cl.logger.Debugf("created filter for %v (namespaceFilter=%v, extraFilter=%v, fieldSelector=%v)", resourceGVK, namespaceFilter != nil, extraFilter != nil, fieldSelector)
204+
if resourceGVK == gvk.KubernetesGateway {
205+
filter.ObjectFilter = kubetypes.ComposeFilters(namespaceFilter, extraFilter)
206+
}
205207

206208
var kc kclient.Untyped
207209
if s.IsBuiltin() {
208210
kc = kclient.NewUntypedInformer(cl.client, gvr, filter)
209211
} else {
210-
// For DestinationRule and VirtualService, we use Dynamic client which returns unstructured objects
211-
// So we need to use DynamicInformer type to ensure the informer expects unstructured objects
212-
informerType := kubetypes.StandardInformer
213-
if resourceGVK == gvk.DestinationRule || resourceGVK == gvk.VirtualService || resourceGVK == gvk.PeerAuthentication {
214-
informerType = kubetypes.DynamicInformer
215-
cl.logger.Debugf("using DynamicInformer for %v (uses Dynamic client)", resourceGVK)
216-
}
217212
kc = kclient.NewDelayedInformer[controllers.Object](
218213
cl.client,
219214
gvr,
220-
informerType,
215+
kubetypes.StandardInformer,
221216
filter,
222217
)
223218
}
@@ -226,58 +221,18 @@ func (cl *Client) addCRD(name string, opts krt.OptionsBuilder) {
226221
collection := krt.MapCollection(wrappedClient, func(obj controllers.Object) config.Config {
227222
cfg := translateFunc(obj)
228223
cfg.Domain = cl.domainSuffix
229-
// Only log at Debug level to avoid spam, but keep it available for diagnosis
230-
cl.logger.Debugf("MapCollection translating object %s/%s to config for %v", obj.GetNamespace(), obj.GetName(), resourceGVK)
231224
return cfg
232225
}, opts.WithName("collection/"+resourceGVK.Kind)...)
233226
index := krt.NewNamespaceIndex(collection)
234-
// Register a debug handler to track all events from the wrappedClient (before MapCollection)
235-
// This helps diagnose if events are being filtered before reaching the collection
236-
wrappedClientDebugHandler := wrappedClient.RegisterBatch(func(o []krt.Event[controllers.Object]) {
237-
if len(o) > 0 {
238-
cl.logger.Debugf("wrappedClient event detected for %v: %d events", resourceGVK, len(o))
239-
for i, event := range o {
240-
var nameStr, nsStr string
241-
if event.New != nil {
242-
obj := *event.New
243-
nameStr = obj.GetName()
244-
nsStr = obj.GetNamespace()
245-
} else if event.Old != nil {
246-
obj := *event.Old
247-
nameStr = obj.GetName()
248-
nsStr = obj.GetNamespace()
249-
}
250-
cl.logger.Debugf("wrappedClient event[%d] %s for %v (name=%s/%s)",
251-
i, event.Event, resourceGVK, nsStr, nameStr)
252-
}
253-
}
254-
}, false)
255-
// Register a debug handler to track all events from the collection
256-
// This helps diagnose why new config changes might not trigger events
257-
// Use false to match Dubbo's implementation - only process future events, not initial sync
258-
debugHandler := collection.RegisterBatch(func(o []krt.Event[config.Config]) {
259-
if len(o) > 0 {
260-
cl.logger.Debugf("collection event detected for %v: %d events", resourceGVK, len(o))
261-
for i, event := range o {
262-
var nameStr, nsStr string
263-
if event.New != nil {
264-
nameStr = event.New.Name
265-
nsStr = event.New.Namespace
266-
} else if event.Old != nil {
267-
nameStr = event.Old.Name
268-
nsStr = event.Old.Namespace
269-
}
270-
cl.logger.Debugf("collection event[%d] %s for %v (name=%s/%s)",
271-
i, event.Event, resourceGVK, nsStr, nameStr)
272-
}
273-
}
274-
}, false)
275227
cl.kinds[resourceGVK] = nsStore{
276228
collection: collection,
277229
index: index,
278230
handlers: []krt.HandlerRegistration{
279-
wrappedClientDebugHandler,
280-
debugHandler,
231+
collection.RegisterBatch(func(o []krt.Event[config.Config]) {
232+
// for _, event := range o {
233+
// incrementEvent(resourceGVK.Kind, event.Event.String())
234+
// }
235+
}, false),
281236
},
282237
}
283238
}
@@ -294,55 +249,23 @@ func (cl *Client) Schemas() collection.Schemas {
294249
}
295250

296251
func (cl *Client) RegisterEventHandler(kind config.GroupVersionKind, handler model.EventHandler) {
297-
cl.kindsMu.Lock()
298-
defer cl.kindsMu.Unlock()
299-
300-
c, ok := cl.kinds[kind]
301-
if !ok {
302-
cl.logger.Warnf("unknown type: %s", kind)
303-
return
304-
}
305-
306-
cl.logger.Debugf("Registering handler for %v", kind)
307-
handlerReg := c.collection.RegisterBatch(func(o []krt.Event[config.Config]) {
308-
cl.logger.Debugf("batch handler triggered for %v with %d events", kind, len(o))
309-
for i, event := range o {
310-
var nameStr, nsStr string
311-
if event.New != nil {
312-
nameStr = event.New.Name
313-
nsStr = event.New.Namespace
314-
} else if event.Old != nil {
315-
nameStr = event.Old.Name
316-
nsStr = event.Old.Namespace
317-
}
318-
cl.logger.Debugf("processing event[%d] %s for %v (name=%s/%s)",
319-
i, event.Event, kind, nsStr, nameStr)
320-
switch event.Event {
321-
case controllers.EventAdd:
322-
if event.New != nil {
252+
if c, ok := cl.kind(kind); ok {
253+
c.handlers = append(c.handlers, c.collection.RegisterBatch(func(o []krt.Event[config.Config]) {
254+
for _, event := range o {
255+
switch event.Event {
256+
case controllers.EventAdd:
323257
handler(config.Config{}, *event.New, model.Event(event.Event))
324-
} else {
325-
cl.logger.Warnf("EventAdd but event.New is nil, skipping")
326-
}
327-
case controllers.EventUpdate:
328-
if event.Old != nil && event.New != nil {
258+
case controllers.EventUpdate:
329259
handler(*event.Old, *event.New, model.Event(event.Event))
330-
} else {
331-
cl.logger.Warnf("EventUpdate but event.Old or event.New is nil, skipping")
332-
}
333-
case controllers.EventDelete:
334-
if event.Old != nil {
260+
case controllers.EventDelete:
335261
handler(config.Config{}, *event.Old, model.Event(event.Event))
336-
} else {
337-
cl.logger.Warnf("EventDelete but event.Old is nil, skipping")
338262
}
339263
}
340-
}
341-
}, false)
342-
// Update handlers slice to keep reference (though not strictly necessary for functionality)
343-
c.handlers = append(c.handlers, handlerReg)
344-
cl.kinds[kind] = c
345-
cl.logger.Debugf("Successfully registered handler for %v", kind)
264+
}, false))
265+
return
266+
}
267+
268+
cl.logger.Warnf("unknown type: %s", kind)
346269
}
347270

348271
func (cl *Client) Get(typ config.GroupVersionKind, name, namespace string) *config.Config {
@@ -421,43 +344,14 @@ func (cl *Client) Delete(typ config.GroupVersionKind, name, namespace string, re
421344
func (cl *Client) List(kind config.GroupVersionKind, namespace string) []config.Config {
422345
h, f := cl.kind(kind)
423346
if !f {
424-
cl.logger.Warnf("unknown kind %v", kind)
425347
return nil
426348
}
427349

428-
// Check if collection is synced
429-
if !h.collection.HasSynced() {
430-
cl.logger.Warnf("collection for %v is not synced yet", kind)
431-
}
432-
433-
var configs []config.Config
434350
if namespace == metav1.NamespaceAll {
435-
// Get all configs from collection
436-
configs = h.collection.List()
437-
cl.logger.Debugf("found %d configs for %v (namespace=all, synced=%v)",
438-
len(configs), kind, h.collection.HasSynced())
439-
if len(configs) > 0 {
440-
for i, cfg := range configs {
441-
cl.logger.Debugf("config[%d] %s/%s for %v", i, cfg.Namespace, cfg.Name, kind)
442-
}
443-
} else {
444-
cl.logger.Debugf("collection returned 0 configs for %v (synced=%v), this may indicate informer is not watching correctly or resources are being filtered", kind, h.collection.HasSynced())
445-
}
446-
// Log collection type for diagnosis
447-
cl.logger.Debugf("collection type is %T, HasSynced=%v", h.collection, h.collection.HasSynced())
448-
} else {
449-
configs = h.index.Lookup(namespace)
450-
cl.logger.Debugf("found %d configs for %v in namespace %s (synced=%v)", len(configs), kind, namespace, h.collection.HasSynced())
451-
if len(configs) > 0 {
452-
for i, cfg := range configs {
453-
cl.logger.Debugf("config[%d] %s/%s for %v", i, cfg.Namespace, cfg.Name, kind)
454-
}
455-
} else {
456-
cl.logger.Debugf("found 0 configs for %v in namespace %s (synced=%v), checking if resources exist in cluster", kind, namespace, h.collection.HasSynced())
457-
}
351+
return h.collection.List()
458352
}
459353

460-
return configs
354+
return h.index.Lookup(namespace)
461355
}
462356

463357
func getObjectMetadata(config config.Config) metav1.ObjectMeta {

go.mod

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -58,13 +58,13 @@ require (
5858
github.com/hashicorp/golang-lru/v2 v2.0.5
5959
github.com/heroku/color v0.0.6
6060
github.com/howardjohn/celpp v0.1.0
61-
github.com/miekg/dns v1.1.68
6261
github.com/moby/term v0.5.2
6362
github.com/ory/viper v1.7.5
6463
github.com/pkg/errors v0.9.1
6564
github.com/sashabaranov/go-openai v1.40.5
6665
github.com/spf13/cobra v1.9.1
6766
github.com/spf13/pflag v1.0.7
67+
github.com/stoewer/go-strcase v1.3.0
6868
github.com/tmc/langchaingo v0.1.13
6969
go.uber.org/atomic v1.11.0
7070
golang.org/x/crypto v0.41.0
@@ -241,7 +241,6 @@ require (
241241
github.com/spf13/afero v1.14.0 // indirect
242242
github.com/spf13/cast v1.8.0 // indirect
243243
github.com/spf13/jwalterweatherman v1.1.0 // indirect
244-
github.com/stoewer/go-strcase v1.3.0 // indirect
245244
github.com/subosito/gotenv v1.6.0 // indirect
246245
github.com/ulikunitz/xz v0.5.12 // indirect
247246
github.com/vbatts/tar-split v0.12.1 // indirect

go.sum

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -484,8 +484,6 @@ github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b h1:j7+1HpAFS1zy5+Q4qx1f
484484
github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b/go.mod h1:01TrycV0kFyexm33Z7vhZRXopbI8J3TDReVlkTgMUxE=
485485
github.com/microcosm-cc/bluemonday v1.0.26 h1:xbqSvqzQMeEHCqMi64VAs4d8uy6Mequs3rQ0k/Khz58=
486486
github.com/microcosm-cc/bluemonday v1.0.26/go.mod h1:JyzOCs9gkyQyjs+6h10UEVSe02CGwkhd72Xdqh78TWs=
487-
github.com/miekg/dns v1.1.68 h1:jsSRkNozw7G/mnmXULynzMNIsgY2dHC8LO6U6Ij2JEA=
488-
github.com/miekg/dns v1.1.68/go.mod h1:fujopn7TB3Pu3JM69XaawiU0wqjpL9/8xGop5UrTPps=
489487
github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw=
490488
github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
491489
github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=

manifests/charts/dubbo-control/dubbo-discovery/files/grpc-agent.yaml

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,8 +38,7 @@ spec:
3838
postStart:
3939
exec:
4040
command:
41-
# - dubbo-agent
42-
- planet-agent
41+
- dubbo-agent
4342
- wait
4443
- --url=http://localhost:15020/healthz/ready
4544
readinessProbe:

manifests/charts/dubbo-control/dubbo-discovery/files/gateway.yaml renamed to manifests/charts/dubbo-control/dubbo-discovery/files/kube-gateway.yaml

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ spec:
4646
serviceAccountName: {{ .ServiceAccount }}
4747
containers:
4848
- name: dubbo-proxy
49-
image: mfordjody/proxyadapter:0.3.2-debug
49+
image: mfordjody/dubbo-proxy:0.3.5
5050
imagePullPolicy: Always
5151
ports:
5252
- containerPort: 15020
@@ -144,6 +144,11 @@ kind: Service
144144
metadata:
145145
name: {{ .DeploymentName }}
146146
namespace: {{ .Namespace }}
147+
ownerReferences:
148+
- apiVersion: gateway.networking.k8s.io/v1
149+
kind: Gateway
150+
name: {{.Name}}
151+
uid: {{.UID}}
147152
labels:
148153
gateway.dubbo.apache.org/managed: {{ .ControllerLabel }}
149154
gateway.networking.k8s.io/gateway-name: {{ .Name }}

manifests/charts/dubbo-control/dubbo-discovery/templates/deployment.yaml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ spec:
1919
serviceAccountName: dubbod
2020
containers:
2121
- name: discovery
22-
image: "mfordjody/planet:0.3.0-debug"
22+
image: "mfordjody/dubbo-discovery:0.3.5"
2323
imagePullPolicy: Always
2424
args:
2525
- "discovery"

manifests/charts/dubbo-control/dubbo-discovery/templates/dubbd-injector-configmap.yaml

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,12 @@ metadata:
88
dubbo.apache.org/rev: default
99
data:
1010
config: |-
11-
defaultTemplates: [grpc-agent]
11+
defaultTemplates: [kube-gateway]
1212
policy: enabled
1313
templates:
1414
grpc-agent: |
1515
{{ .Files.Get "files/grpc-agent.yaml" | trim | indent 8 }}
1616
gateway: |
17-
{{ .Files.Get "files/gateway.yaml" | trim | indent 8 }}
17+
{{ .Files.Get "files/kube-gateway.yaml" | trim | indent 8 }}
1818
values: |-
1919

manifests/charts/dubbo-control/dubbo-discovery/values.yaml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515

1616
_internal_default_values_not_set:
1717
hub: ""
18-
tag: 0.3.0
18+
tag: 0.3.5
1919
image: planet
2020

2121
resources:

0 commit comments

Comments
 (0)