| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package subscriber |
| |
| import ( |
| "errors" |
| "reflect" |
| |
| "github.com/duke-git/lancet/v2/slice" |
| "github.com/duke-git/lancet/v2/strutil" |
| "k8s.io/client-go/tools/cache" |
| |
| "github.com/apache/dubbo-admin/pkg/common/bizerror" |
| "github.com/apache/dubbo-admin/pkg/common/constants" |
| enginecfg "github.com/apache/dubbo-admin/pkg/config/engine" |
| "github.com/apache/dubbo-admin/pkg/core/events" |
| "github.com/apache/dubbo-admin/pkg/core/logger" |
| meshresource "github.com/apache/dubbo-admin/pkg/core/resource/apis/mesh/v1alpha1" |
| coremodel "github.com/apache/dubbo-admin/pkg/core/resource/model" |
| "github.com/apache/dubbo-admin/pkg/core/store" |
| "github.com/apache/dubbo-admin/pkg/core/store/index" |
| ) |
| |
| type RuntimeInstanceEventSubscriber struct { |
| instanceStore store.ResourceStore |
| eventEmitter events.Emitter |
| } |
| |
| func NewRuntimeInstanceEventSubscriber(instanceResourceStore store.ResourceStore, emitter events.Emitter) events.Subscriber { |
| return &RuntimeInstanceEventSubscriber{ |
| instanceStore: instanceResourceStore, |
| eventEmitter: emitter, |
| } |
| } |
| |
| func (s *RuntimeInstanceEventSubscriber) ResourceKind() coremodel.ResourceKind { |
| return meshresource.RuntimeInstanceKind |
| } |
| |
| func (s *RuntimeInstanceEventSubscriber) Name() string { |
| return "Engine-" + s.ResourceKind().ToString() |
| } |
| |
| func (s *RuntimeInstanceEventSubscriber) ProcessEvent(event events.Event) error { |
| newObj, ok := event.NewObj().(*meshresource.RuntimeInstanceResource) |
| if !ok && event.NewObj() != nil { |
| return bizerror.NewAssertionError(meshresource.RuntimeInstanceKind, reflect.TypeOf(event.NewObj()).Name()) |
| } |
| oldObj, ok := event.OldObj().(*meshresource.RuntimeInstanceResource) |
| if !ok && event.OldObj() != nil { |
| return bizerror.NewAssertionError(meshresource.RuntimeInstanceKind, reflect.TypeOf(event.OldObj()).Name()) |
| } |
| var processErr error |
| switch event.Type() { |
| case cache.Added, cache.Updated, cache.Replaced, cache.Sync: |
| if newObj == nil { |
| errStr := "process runtime instance upsert event, but new obj is nil, skipped processing" |
| logger.Error(errStr) |
| return errors.New(errStr) |
| } |
| processErr = s.processUpsert(newObj) |
| case cache.Deleted: |
| if oldObj == nil { |
| errStr := "process runtime instance delete event, but old obj is nil, skipped processing" |
| logger.Error(errStr) |
| return errors.New(errStr) |
| } |
| processErr = s.processDelete(oldObj) |
| } |
| if processErr != nil { |
| logger.Errorf("process runtime instance event failed, cause: %s event: %s", processErr.Error(), event.String()) |
| return processErr |
| } |
| logger.Infof("process runtime instance event successfully, event: %s", event.String()) |
| return nil |
| } |
| |
| // processUpsert when runtime instance added or updated, we should add/update the corresponding instance resource |
| func (s *RuntimeInstanceEventSubscriber) processUpsert(rtInstanceRes *meshresource.RuntimeInstanceResource) error { |
| var instanceResource *meshresource.InstanceResource |
| var err error |
| switch rtInstanceRes.Spec.SourceEngineType { |
| case string(enginecfg.Kubernetes): |
| instanceResource, err = s.getRelatedInstanceByIP(rtInstanceRes) |
| default: |
| instanceResource, err = s.getRelatedInstanceByName(rtInstanceRes) |
| } |
| if err != nil { |
| return err |
| } |
| // if instance resource exists, that is to say the rpc instance exists in remote registry and has been watched by discovery. |
| // so we should merge the runtime info into it |
| if instanceResource != nil { |
| meshresource.MergeRuntimeInstanceIntoInstance(rtInstanceRes, instanceResource) |
| return s.instanceStore.Update(instanceResource) |
| } |
| // if instance resource does not exist, that is to say the rpc instance does not exist in remote registry. |
| // we need to check whether the runtime instance resource is enough to create a new instance resource |
| if !checkAttributesEnough(rtInstanceRes) { |
| logger.Warnf("cannot identify runtime instance %s to a dubbo instance, skipped creating new instance", rtInstanceRes.ResourceKey()) |
| return nil |
| } |
| // if conditions met, we should create a new instance resource by runtime instance |
| instanceRes := meshresource.FromRuntimeInstance(rtInstanceRes) |
| if err = s.instanceStore.Add(instanceRes); err != nil { |
| logger.Errorf("add instance resource failed, instance: %s, err: %s", instanceRes.ResourceKey(), err.Error()) |
| return err |
| } |
| instanceAddEvent := events.NewResourceChangedEvent(cache.Added, nil, instanceRes) |
| s.eventEmitter.Send(instanceAddEvent) |
| logger.Debugf("runtime instance upsert trigger instance add event, event: %s", instanceAddEvent.String()) |
| return nil |
| } |
| |
| // processDelete when runtime instance deleted, we should delete the corresponding instance resource |
| func (s *RuntimeInstanceEventSubscriber) processDelete(rtInstanceRes *meshresource.RuntimeInstanceResource) error { |
| var instanceResource *meshresource.InstanceResource |
| var err error |
| switch rtInstanceRes.Spec.SourceEngineType { |
| case string(enginecfg.Kubernetes): |
| instanceResource, err = s.getRelatedInstanceByIP(rtInstanceRes) |
| default: |
| instanceResource, err = s.getRelatedInstanceByName(rtInstanceRes) |
| } |
| if err != nil { |
| return err |
| } |
| if instanceResource == nil { |
| logger.Warnf("cannot find instance resource by runtime instance %s, skipped deleting instance", rtInstanceRes.ResourceKey()) |
| return nil |
| } |
| if err = s.instanceStore.Delete(instanceResource); err != nil { |
| logger.Errorf("delete instance resource failed, instance: %s, err: %s", instanceResource.ResourceKey(), err.Error()) |
| return err |
| } |
| instanceDeleteEvent := events.NewResourceChangedEvent(cache.Deleted, instanceResource, nil) |
| s.eventEmitter.Send(instanceDeleteEvent) |
| logger.Debugf("runtime instance delete trigger instance delete event, event: %s", instanceDeleteEvent.String()) |
| return nil |
| } |
| |
| func (s *RuntimeInstanceEventSubscriber) getRelatedInstanceByName( |
| rtInstanceRes *meshresource.RuntimeInstanceResource) (*meshresource.InstanceResource, error) { |
| if rtInstanceRes.Spec == nil || strutil.IsBlank(rtInstanceRes.Spec.AppName) || |
| strutil.IsBlank(rtInstanceRes.Spec.Ip) || rtInstanceRes.Spec.RpcPort <= 0 { |
| return nil, nil |
| } |
| instanceResName := meshresource.BuildInstanceResName(rtInstanceRes.Spec.AppName, rtInstanceRes.Spec.Ip, rtInstanceRes.Spec.RpcPort) |
| resources, err := s.instanceStore.ListByIndexes(map[string]string{ |
| index.ByInstanceNameIndex: instanceResName, |
| }) |
| if err != nil { |
| return nil, err |
| } |
| if len(resources) == 0 { |
| return nil, nil |
| } |
| instanceResList := make([]*meshresource.InstanceResource, len(resources)) |
| for i, item := range resources { |
| res, ok := item.(*meshresource.InstanceResource) |
| if !ok { |
| return nil, bizerror.NewAssertionError("InstanceResource", reflect.TypeOf(item).Name()) |
| } |
| instanceResList[i] = res |
| } |
| if len(instanceResList) > 1 { |
| resKeys := slice.Map(instanceResList, func(index int, item *meshresource.InstanceResource) string { |
| return item.ResourceKey() |
| }) |
| logger.Warnf("there are more than two instances which have the same name, instance keys: %s, name: %s", resKeys, instanceResName) |
| return nil, nil |
| } |
| return instanceResList[0], nil |
| } |
| |
| func (s *RuntimeInstanceEventSubscriber) getRelatedInstanceByIP( |
| rtInstanceRes *meshresource.RuntimeInstanceResource) (*meshresource.InstanceResource, error) { |
| resources, err := s.instanceStore.ListByIndexes(map[string]string{ |
| index.ByInstanceIpIndex: rtInstanceRes.Spec.Ip, |
| }) |
| if err != nil { |
| return nil, err |
| } |
| if len(resources) == 0 { |
| return nil, nil |
| } |
| instanceResList := make([]*meshresource.InstanceResource, len(resources)) |
| for i, item := range resources { |
| res, ok := item.(*meshresource.InstanceResource) |
| if !ok { |
| return nil, bizerror.NewAssertionError("InstanceResource", reflect.TypeOf(item).Name()) |
| } |
| instanceResList[i] = res |
| } |
| if len(instanceResList) > 1 { |
| resKeys := slice.Map(instanceResList, func(index int, item *meshresource.InstanceResource) string { |
| return item.ResourceKey() |
| }) |
| logger.Warnf("there are instances which have same ip, instance keys: %s, ip: %s", resKeys, rtInstanceRes.Spec.Ip) |
| return nil, nil |
| } |
| return instanceResList[0], nil |
| } |
| |
| func checkAttributesEnough(rtInstanceRes *meshresource.RuntimeInstanceResource) bool { |
| if rtInstanceRes.Spec == nil || strutil.IsBlank(rtInstanceRes.Spec.AppName) || |
| strutil.IsBlank(rtInstanceRes.Spec.Ip) || rtInstanceRes.Spec.RpcPort <= 0 || |
| strutil.IsBlank(rtInstanceRes.Mesh) || constants.DefaultMesh == rtInstanceRes.Mesh { |
| return false |
| } |
| return true |
| } |
