如何为淮安地区创建专业的Joomla网站并进行新网站的建设？

joomla 网站建设教程,淮安新网站制作,mediwiki 做网站,网站页面设计需求文档apisix Apache APISIX 是一个基于微服务 API 网关#xff0c;其不仅可以处理南北向的流量#xff0c;也可以处理东西向的流量即服务之间的流量。Apache APISIX 集成了控制面板和数据面#xff0c;与其他 API 网关相比#xff0c;Apache APISIX 的上游、路由、插件全是动态的… apisix Apache APISIX 是一个基于微服务 API 网关其不仅可以处理南北向的流量也可以处理东西向的流量即服务之间的流量。Apache APISIX 集成了控制面板和数据面与其他 API 网关相比Apache APISIX 的上游、路由、插件全是动态的修改这些东西时都不用重启。并且 Apache APISIX 的插件也是热加载可以随时插拔、修改插件。 Apache APISIX 其设计理念是基于API 网关的数据平面和控制平面分离。控制平面不仅仅能够控制 Apache APISIX 同时其还能够控制其他组件数据平面不仅仅能够被自身的控制平面控制还能被其他组件所控制。由于其基于ETCD 来存储和分发路由数据默认具备高可用无单点故障风险。除此之外其能够友好地支持 Prometheus、SkyWalking 动态追踪、流量复制、故障注入等相关功能。 apisix ingress 在 K8s 生态中Ingress 作为表示 K8s 流量入口的一种资源想要让其生效就需要有一个 Ingress Controller 去监听 K8s 中的 Ingress 资源并对这些资源进行相应规则的解析和实际承载流量。在当下趋势中像 Kubernetes Ingress Nginx 就是使用最广泛的 Ingress Controller 实现。 而 APISIX Ingress 则是另一种 Ingress Controller 的实现。跟 Kubernetes Ingress Nginx 的区别主要在于 APISIX Ingress 是以 Apache APISIX 作为实际承载业务流量的数据面。 Apache APISIX Ingress Controller 除了覆盖 NGINX Ingress Controller 已有的能力外还解决了一些 Nginx Ingress Controller 的痛点。具体如下 1、配置的动态化加载通常情况下作为接入层的 Ingress Controller 其承载着服务的入口流量引入在生产环境中我们的业务对系统的可靠性有着更高的要求然而基于 Apache APISIX Ingress Controller 其能够支持动态配置即时生效降低生产事故的意外及风险有助于提高运维可维护性。2、较强的灰度能力在实际的业务场景中有的时候往往会依据某些特定的需求进行权重调整结合业务需求按比例进行流量控制Apache APISIX Ingress Controller 可以支持 Service和 Pod 级别的权重调整配置清晰而且可读性更强。除此相对于NGINX Ingress Controller 中通过 Annotation 的方式提供 Canary 灰度方案Apache APISIX Ingress Controller 能够解决其缺陷从而能够更好的提供灰度策略。3、较好的扩展能力基于 Apache APISIX 强大的插件能力Apache APISIX Ingress Controller 通过动态绑定插件来增强功能。Apache APISIX 通过插件封装逻辑易于管理完善的文档易于使用和理解。Apache APISIX Ingress Controller 通过配置即可绑定和解绑插件无需操作脚本。 APISIX Ingress 目前已经支持的自定义资源主要是以下 5 类涉及到路由、上游、消费者、证书相关和集群公共配置的相关类别。 内置服务发现 APISIX 内置了下面这些服务发现机制 基于 Eureka 的服务发现基于 Nacos 的服务发现基于 Consul 的服务发现基于 Consul KV 的服务发现基于 DNS 的服务发现基于 APISIX-Seed 架构的控制面服务发现基于 Kubernetes 的服务发现 上面介绍的这些都是基于数据面apisix配置信息手动变更操作的方案集成。其实在基于k8s的云原生场景下apisix还提供了一个控制面组件来对apisix的服务发现进行自动管理那就是apisix ingress。 下面就从源码的角度来看看apisix ingress是怎么做到自动的服务发现的 apisix ingress启动 从main.go启动入口一路跟踪进入providers/controller.go的run方法: func (c *Controller) run(ctx context.Context) {log.Infow(controller tries to leading ...,zap.String(namespace, c.namespace),zap.String(pod, c.name),)var cancelFunc context.CancelFuncctx, cancelFunc context.WithCancel(ctx)defer cancelFunc()// give up leaderdefer c.leaderContextCancelFunc()clusterOpts : apisix.ClusterOptions{AdminAPIVersion: c.cfg.APISIX.AdminAPIVersion,Name: c.cfg.APISIX.DefaultClusterName,AdminKey: c.cfg.APISIX.DefaultClusterAdminKey,BaseURL: c.cfg.APISIX.DefaultClusterBaseURL,MetricsCollector: c.MetricsCollector,}err : c.apisix.AddCluster(ctx, clusterOpts)if err ! nil  err ! apisix.ErrDuplicatedCluster {// TODO give up the leader rolelog.Errorf(failed to add default cluster: %s, err)return}if err : c.apisix.Cluster(c.cfg.APISIX.DefaultClusterName).HasSynced(ctx); err ! nil {// TODO give up the leader rolelog.Errorf(failed to wait the default cluster to be ready: %s, err)// re-create apisix cluster, used in next c.runif err c.apisix.UpdateCluster(ctx, clusterOpts); err ! nil {log.Errorf(failed to update default cluster: %s, err)return}return}// Creation Phasec.informers c.initSharedInformers()common : providertypes.Common{ControllerNamespace: c.namespace,ListerInformer: c.informers,Config: c.cfg,APISIX: c.apisix,KubeClient: c.kubeClient,MetricsCollector: c.MetricsCollector,Recorder: c.recorder,}c.namespaceProvider, err namespace.NewWatchingNamespaceProvider(ctx, c.kubeClient, c.cfg)if err ! nil {ctx.Done()return}c.podProvider, err pod.NewProvider(common, c.namespaceProvider)if err ! nil {ctx.Done()return}c.translator translation.NewTranslator(translation.TranslatorOptions{APIVersion: c.cfg.Kubernetes.APIVersion,EndpointLister: c.informers.EpLister,ServiceLister: c.informers.SvcLister,SecretLister: c.informers.SecretLister,PodLister: c.informers.PodLister,ApisixUpstreamLister: c.informers.ApisixUpstreamLister,PodProvider: c.podProvider,})c.apisixProvider, c.apisixTranslator, err apisixprovider.NewProvider(common, c.namespaceProvider, c.translator)if err ! nil {ctx.Done()return}c.ingressProvider, err ingressprovider.NewProvider(common, c.namespaceProvider, c.translator, c.apisixTranslator)if err ! nil {ctx.Done()return}c.kubeProvider, err k8s.NewProvider(common, c.translator, c.namespaceProvider, c.apisixProvider, c.ingressProvider)if err ! nil {ctx.Done()return}if c.cfg.Kubernetes.EnableGatewayAPI {c.gatewayProvider, err gateway.NewGatewayProvider(gateway.ProviderOptions{Cfg: c.cfg,APISIX: c.apisix,APISIXClusterName: c.cfg.APISIX.DefaultClusterName,KubeTranslator: c.translator,RestConfig: nil,KubeClient: c.kubeClient.Client,MetricsCollector: c.MetricsCollector,NamespaceProvider: c.namespaceProvider,})if err ! nil {ctx.Done()return}}// Init Phaseif err c.namespaceProvider.Init(ctx); err ! nil {ctx.Done()return}if err c.apisixProvider.Init(ctx); err ! nil {ctx.Done()return}// Run Phasee : utils.ParallelExecutor{}e.Add(func() {c.checkClusterHealth(ctx, cancelFunc)})e.Add(func() {c.informers.Run(ctx)})e.Add(func() {c.namespaceProvider.Run(ctx)})e.Add(func() {c.kubeProvider.Run(ctx)})e.Add(func() {c.apisixProvider.Run(ctx)})e.Add(func() {c.ingressProvider.Run(ctx)})if c.cfg.Kubernetes.EnableGatewayAPI {e.Add(func() {c.gatewayProvider.Run(ctx)})}e.Add(func() {c.resourceSyncLoop(ctx, c.cfg.ApisixResourceSyncInterval.Duration)})c.MetricsCollector.ResetLeader(true)log.Infow(controller now is running as leader,zap.String(namespace, c.namespace),zap.String(pod, c.name),)-ctx.Done()e.Wait()for _, execErr : range e.Errors() {log.Error(execErr.Error())}if len(e.Errors()) 0 {log.Error(Start failed, abort...)cancelFunc()} } 上面代码逻辑大致如下 初始化apisix集群配置信息用于跟数据面服务apisix通信进行相关的配置操作初始化k8s资源inforrmers信息用户对k8s各个资源进行监听及获取资源信息监听k8s集群namespace资源并处理监听k8s集群pod资源并处理监听k8s集群ingress资源并处理监听k8s集群中apisix自定义资源并处理比如apisixRoute等监听k8s集群endpoint资源并处理监听k8s集群secret资源并处理监听k8s集群configmap资源并处理监听k8s集群gateway资源并处理 下面以处理endpoint资源为例进行说明其他资源的监听处理类似就不一一讲解了。 服务发现 进入到k8s/endpoint/provider.go中我们先来看看实例初始化方法 func NewProvider(common *providertypes.Common, translator translation.Translator, namespaceProvider namespace.WatchingNamespaceProvider) (Provider, error) {p : endpointProvider{cfg: common.Config,}base : baseEndpointController{Common: common,translator: translator,svcLister: common.SvcLister,apisixUpstreamLister: common.ApisixUpstreamLister,}if common.Kubernetes.WatchEndpointSlices {p.endpointSliceController newEndpointSliceController(base, namespaceProvider)} else {p.endpointsController newEndpointsController(base, namespaceProvider)}return p, nil } func newEndpointsController(base *baseEndpointController, namespaceProvider namespace.WatchingNamespaceProvider) *endpointsController {ctl : endpointsController{baseEndpointController: base,workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.NewItemFastSlowRateLimiter(1*time.Second, 60*time.Second, 5), endpoints),workers: 1,namespaceProvider: namespaceProvider,epLister: base.EpLister,epInformer: base.EpInformer,}ctl.epInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{AddFunc: ctl.onAdd,UpdateFunc: ctl.onUpdate,DeleteFunc: ctl.onDelete,},)return ctl } 注意到最后代码的AddEventHandler这里就是我们经常见到的informer的处理回调方法设置的地方。 我们看到针对endpoint资源的增删改设置了对应的回调方法。这里来看看onAdd方法 func (c *endpointsController) onAdd(obj interface{}) {key, err : cache.MetaNamespaceKeyFunc(obj)if err ! nil {log.Errorf(found endpoints object with bad namespace/name: %s, ignore it, err)return}if !c.namespaceProvider.IsWatchingNamespace(key) {return}log.Debugw(endpoints add event arrived,zap.String(object-key, key))c.workqueue.Add(types.Event{Type: types.EventAdd,// TODO pass key.Object: kube.NewEndpoint(obj.(*corev1.Endpoints)),})c.MetricsCollector.IncrEvents(endpoints, add) } 该方法的参数表示增加的endpoint资源对象信息。该方法主要是向endpoint的队列workqueue中增加一个事件对象包含事件类型、增加的endpoint对象。 在最开始main我们介绍provider的启动方法中提到执行了每个provider的run方法下面我们来看下endpoint的provider的run方法 func (c *endpointsController) run(ctx context.Context) {log.Info(endpoints controller started)defer log.Info(endpoints controller exited)defer c.workqueue.ShutDown()if ok : cache.WaitForCacheSync(ctx.Done(), c.epInformer.HasSynced); !ok {log.Error(informers sync failed)return}handler : func() {for {obj, shutdown : c.workqueue.Get()if shutdown {return}err : c.sync(ctx, obj.(*types.Event))c.workqueue.Done(obj)c.handleSyncErr(obj, err)}}for i : 0; i c.workers; i {go handler()}-ctx.Done() } func (c *endpointsController) sync(ctx context.Context, ev *types.Event) error {ep : ev.Object.(kube.Endpoint)ns, err : ep.Namespace()if err ! nil {return err}newestEp, err : c.epLister.GetEndpoint(ns, ep.ServiceName())if err ! nil {if errors.IsNotFound(err) {return c.syncEmptyEndpoint(ctx, ep)}return err}return c.syncEndpoint(ctx, newestEp) } func (c *baseEndpointController) syncEndpoint(ctx context.Context, ep kube.Endpoint) error {log.Debugw(endpoint controller syncing endpoint,zap.Any(endpoint, ep),)namespace, err : ep.Namespace()if err ! nil {return err}svcName : ep.ServiceName()svc, err : c.svcLister.Services(namespace).Get(svcName)if err ! nil {if k8serrors.IsNotFound(err) {return c.syncEmptyEndpoint(ctx, ep)}log.Errorf(failed to get service %s/%s: %s, namespace, svcName, err)return err}switch c.Kubernetes.APIVersion {case config.ApisixV2beta3:var subsets []configv2beta3.ApisixUpstreamSubsetsubsets append(subsets, configv2beta3.ApisixUpstreamSubset{})auKube, err : c.apisixUpstreamLister.V2beta3(namespace, svcName)if err ! nil {if !k8serrors.IsNotFound(err) {log.Errorf(failed to get ApisixUpstream %s/%s: %s, namespace, svcName, err)return err}} else if auKube.V2beta3().Spec ! nil  len(auKube.V2beta3().Spec.Subsets) 0 {subsets append(subsets, auKube.V2beta3().Spec.Subsets...)}clusters : c.APISIX.ListClusters()for _, port : range svc.Spec.Ports {for _, subset : range subsets {nodes, err : c.translator.TranslateEndpoint(ep, port.Port, subset.Labels)if err ! nil {log.Errorw(failed to translate upstream nodes,zap.Error(err),zap.Any(endpoints, ep),zap.Int32(port, port.Port),)}name : apisixv1.ComposeUpstreamName(namespace, svcName, subset.Name, port.Port, types.ResolveGranularity.Endpoint)for _, cluster : range clusters {if err : c.SyncUpstreamNodesChangeToCluster(ctx, cluster, nodes, name); err ! nil {return err}}}}case config.ApisixV2:var subsets []configv2.ApisixUpstreamSubsetsubsets append(subsets, configv2.ApisixUpstreamSubset{})auKube, err : c.apisixUpstreamLister.V2(namespace, svcName)if err ! nil {if !k8serrors.IsNotFound(err) {log.Errorf(failed to get ApisixUpstream %s/%s: %s, namespace, svcName, err)return err}} else if auKube.V2().Spec ! nil  len(auKube.V2().Spec.Subsets) 0 {subsets append(subsets, auKube.V2().Spec.Subsets...)}clusters : c.APISIX.ListClusters()for _, port : range svc.Spec.Ports {for _, subset : range subsets {nodes, err : c.translator.TranslateEndpoint(ep, port.Port, subset.Labels)if err ! nil {log.Errorw(failed to translate upstream nodes,zap.Error(err),zap.Any(endpoints, ep),zap.Int32(port, port.Port),)}name : apisixv1.ComposeUpstreamName(namespace, svcName, subset.Name, port.Port, types.ResolveGranularity.Endpoint)for _, cluster : range clusters {if err : c.SyncUpstreamNodesChangeToCluster(ctx, cluster, nodes, name); err ! nil {return err}}}}default:panic(fmt.Errorf(unsupported ApisixUpstream version %v, c.Kubernetes.APIVersion))}return nil } 上面代码主要逻辑就是 从endpoint的队列workqueue中获取事件对象根据endpoint信息从k8s集群中获取最新的namespace和service等信息根据namespace和servicename从k8s集群中获取apisix upstream资源信息对每一个service端口向数据面服务apisix发送配置更新请求 https://xiaorui.cc/archives/7369