mirror of
https://github.com/tinyauthapp/tinyauth.git
synced 2026-09-23 11:13:32 +08:00
feat: register crd in kube watcher
This commit is contained in:
@@ -22,6 +22,7 @@ func NewKubernetesCRDExtractor(i KubernetesCRDInput) *KubernetesCRDExtractor {
|
||||
|
||||
func (k *KubernetesCRDExtractor) Extract(app *v1alpha1.Application) ExtractionResult {
|
||||
meta := &ResourceMeta{
|
||||
Typ: ResourceTypeCRD,
|
||||
Name: app.GetName(),
|
||||
Namespace: app.GetNamespace(),
|
||||
}
|
||||
@@ -49,7 +50,7 @@ func (k *KubernetesCRDExtractor) Extract(app *v1alpha1.Application) ExtractionRe
|
||||
|
||||
return ExtractionResult{
|
||||
Meta: meta,
|
||||
Apps: &map[string]model.App{
|
||||
Apps: map[string]model.App{
|
||||
// Convert the CRD to the internal representation
|
||||
meta.Name: app.Spec.ToInternalApp(),
|
||||
},
|
||||
|
||||
@@ -76,6 +76,7 @@ func (k *KubernetesIngressExtractor) getHosts(rules []networking.IngressRule) []
|
||||
|
||||
func (k *KubernetesIngressExtractor) Extract(ingress *networking.Ingress) ExtractionResult {
|
||||
meta := &ResourceMeta{
|
||||
Typ: ResourceTypeIngress,
|
||||
Name: ingress.GetName(),
|
||||
Namespace: ingress.GetNamespace(),
|
||||
}
|
||||
@@ -128,6 +129,6 @@ func (k *KubernetesIngressExtractor) Extract(ingress *networking.Ingress) Extrac
|
||||
|
||||
return ExtractionResult{
|
||||
Meta: meta,
|
||||
Apps: &apps,
|
||||
Apps: apps,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -35,13 +35,14 @@ func ensureResourceMeta(meta *ResourceMeta) bool {
|
||||
}
|
||||
|
||||
type ResourceMeta struct {
|
||||
Typ ResourceType
|
||||
Name string
|
||||
Namespace string
|
||||
}
|
||||
|
||||
type ExtractionResult struct {
|
||||
Meta *ResourceMeta
|
||||
Apps *map[string]model.App
|
||||
Apps map[string]model.App
|
||||
}
|
||||
|
||||
type ResourceType string
|
||||
@@ -60,6 +61,13 @@ var supportedResources = []watchedResource{
|
||||
},
|
||||
typ: ResourceTypeIngress,
|
||||
},
|
||||
{
|
||||
gvr: schema.GroupVersionResource{
|
||||
Group: "tinyauth.app",
|
||||
Version: "v1alpha1",
|
||||
Resource: "applications",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
type typedItem struct {
|
||||
@@ -89,6 +97,15 @@ func (ti *typedItem) fromUnstructured(typ ResourceType, obj *unstructured.Unstru
|
||||
typ: ResourceTypeIngress,
|
||||
ingress: typed,
|
||||
}, nil
|
||||
case ResourceTypeCRD:
|
||||
typed, err := convertFromUnstructured[v1alpha1.Application](obj)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &typedItem{
|
||||
typ: ResourceTypeCRD,
|
||||
crd: typed,
|
||||
}, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unknown resource type %s", typ)
|
||||
}
|
||||
@@ -163,7 +180,7 @@ func NewKubernetesService(i KubernetesServiceInput) (*KubernetesService, error)
|
||||
func (k *KubernetesService) addResource(result ExtractionResult) {
|
||||
k.mu.Lock()
|
||||
defer k.mu.Unlock()
|
||||
k.apps[*result.Meta] = *result.Apps
|
||||
k.apps[*result.Meta] = result.Apps
|
||||
}
|
||||
|
||||
func (k *KubernetesService) removeResource(meta ResourceMeta) {
|
||||
@@ -185,7 +202,7 @@ func (k *KubernetesService) getEntry(locator func(name string, app *model.App) b
|
||||
}
|
||||
}
|
||||
|
||||
func (k *KubernetesService) updateFromItem(res watchedResource, typedItem *typedItem) {
|
||||
func (k *KubernetesService) watchedItemChange(res watchedResource, typedItem *typedItem, event watch.EventType) {
|
||||
if typedItem == nil {
|
||||
k.log.App.Warn().Str("res", res.pretty()).Msg("Resource is nil, skipping")
|
||||
return
|
||||
@@ -214,6 +231,13 @@ func (k *KubernetesService) updateFromItem(res watchedResource, typedItem *typed
|
||||
result = extractor.Extract(typedItem.crd)
|
||||
}
|
||||
|
||||
if event == watch.Deleted {
|
||||
if result.Meta != nil {
|
||||
k.removeResource(*result.Meta)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if result.Apps == nil {
|
||||
k.log.App.Warn().Str("res", res.pretty()).Msg("Failed to extract resource, skipping")
|
||||
if result.Meta != nil {
|
||||
@@ -240,7 +264,7 @@ func (k *KubernetesService) resyncGVR(res watchedResource, ctx context.Context)
|
||||
k.log.App.Warn().Err(err).Str("res", res.pretty()).Msg("Failed to decode resource, skipping")
|
||||
continue
|
||||
}
|
||||
k.updateFromItem(res, newTypedItem)
|
||||
k.watchedItemChange(res, newTypedItem, watch.Modified)
|
||||
}
|
||||
k.log.App.Debug().Str("res", res.pretty()).Int("count", len(list.Items)).Msg("Resync complete")
|
||||
return nil
|
||||
@@ -271,7 +295,7 @@ func (k *KubernetesService) runWatcher(res watchedResource, w watch.Interface, r
|
||||
}
|
||||
switch event.Type {
|
||||
case watch.Added, watch.Modified, watch.Deleted:
|
||||
k.updateFromItem(res, newTypedItem)
|
||||
k.watchedItemChange(res, newTypedItem, event.Type)
|
||||
}
|
||||
case <-resyncTicker.C:
|
||||
if err := k.resyncGVR(res, ctx); err != nil {
|
||||
|
||||
@@ -33,8 +33,6 @@ type ApplicationSpec struct {
|
||||
|
||||
// AppConfig specifies configuration for the application
|
||||
type AppConfig struct {
|
||||
// +required
|
||||
Name string `json:"name,omitempty"`
|
||||
// +required
|
||||
Domain string `json:"domain,omitempty"`
|
||||
}
|
||||
|
||||
@@ -19,7 +19,6 @@ var (
|
||||
func addKnownTypes(scheme *runtime.Scheme) error {
|
||||
scheme.AddKnownTypes(SchemeGroupVersion,
|
||||
&Application{},
|
||||
&ApplicationSet{},
|
||||
)
|
||||
|
||||
metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
|
||||
|
||||
Reference in New Issue
Block a user