idou老师教你学Istio 27:解读Mixer Report流程

网友投稿 487 2022-05-30

1、概述

1、概述

Mixer是Istio的核心组件,提供了遥测数据收集的功能,能够实时采集服务的请求状态等信息,以达到监控服务状态目的。

1.1 核心功能

•前置检查(Check):某服务接收并响应外部请求前,先通过Envoy向Mixer(Policy组件)发送Check请求,做一些access检查,同时确认adaptor所需cache字段,供之后Report接口使用;

•配额管理(Quota):通过配额管理机制,处理多请求时发生的资源竞争;

•遥测数据上报(Report):该服务请求处理结束后,将请求相关的日志,监控等数据,通过Envoy上报给Mixer(telemetry)

1.2 示例图

2、代码分析

2.1 Report代码分析

本节主要介绍Report的详细流程(基于Istio release1.0.0版本,commit id为3a136c90)。Report是mixer server的一个接口,供Envoy通过grpc调用。首先,我们从mixer server的启动入口main函数看起:

func main() {    rootCmd := cmd.GetRootCmd(os.Args[1:], supportedTemplates(), supportedAdapters(), shared.Printf, shared.Fatalf)    if err := rootCmd.Execute(); err != nil {       os.Exit(-1)    } }

idou老师教你学Istio 27:解读Mixer Report流程

在rootCmd中,mixs通过server命令启动了mixer server,从而触发了runserver函数,在runserver中初始化(New)了一个server,我们一起看下newServer的函数,在这个函数中,与我们本节相关的内容就是Mixer初始化了一个grpc服务器NewGRPCServer。

rootCmd.AddCommand(serverCmd(info, adapters, printf, fatalf)) func serverCmd(info map[string]template.Info, adapters []adapter.InfoFn, printf, fatalf shared.FormatFn) *cobra.Command {    sa := server.DefaultArgs()    sa.Templates = info    sa.Adapters = adapters    serverCmd := &cobra.Command{       Use:   "server",       Short: "Starts Mixer as a server",       Run: func(cmd *cobra.Command, args []string) {          runServer(sa, printf, fatalf)       },    }… … } func newServer(a *Args, p *patchTable) (*Server, error) {    grpc.EnableTracing = a.EnableGRPCTracing    s.server = grpc.NewServer(grpcOptions...)    mixerpb.RegisterMixerServer(s.server, api.NewGRPCServer(s.dispatcher, s.gp, s.checkCache)) }

rootC

在这个grpc的服务端中,定义了一个Report接口,这就是我们这节课主要关注的内容(可以看到Check接口也在此定义,我们下节再讲)

func (s *grpcServer) Report(ctx context.Context, req *mixerpb.ReportRequest) (*mixerpb.ReportResponse, error) {    lg.Debugf("Report (Count: %d)", len(req.Attributes))    // 校验attribute是否为空,空则直接return    if len(req.Attributes) == 0 {       return reportResp, nil    }    // 若属性word为空,赋为默认值    for i := 0; i < len(req.Attributes); i++ {       iflen(req.Attributes[i].Words) == 0 {          req.Attributes[i].Words = req.DefaultWords       }    }    // 根据第一条attribute,生成proto包,proto包能跟踪一组attributes    protoBag := attribute.NewProtoBag(&req.Attributes[0], s.globalDict, s.globalWordList)    // 初始化,开始跟踪attributes各个条目中属性    accumBag := attribute.GetMutableBag(protoBag)    // 保存accumBag的增量状态    reportBag := attribute.GetMutableBag(accumBag)    reportSpan, reportCtx := opentracing.StartSpanFromContext(ctx, "Report")    reporter := s.dispatcher.GetReporter(reportCtx)    var errors *multierror.Error    for i := 0; i < len(req.Attributes); i++ {       span, newctx := opentracing.StartSpanFromContext(reportCtx, fmt.Sprintf("attribute bag %d", i))       // 第一个属性已经在创建proto包时创建,在此追踪所有attributes       if i > 0 {          if err := accumBag.UpdateBagFromProto(&req.Attributes[i], s.globalWordList); err != nil {             err = fmt.Errorf("request could not be processed due to invalid attributes: %v", err)             span.LogFields(otlog.String("error", err.Error()))             span.Finish()             errors = multierror.Append(errors, err)             break          }       }       lg.Debug("Dispatching Preprocess")       // 真正开始分发,预处理阶段       if err := s.dispatcher.Preprocess(newctx, accumBag, reportBag); err != nil {          err = fmt.Errorf("preprocessing attributes failed: %v", err)          span.LogFields(otlog.String("error", err.Error()))          span.Finish()          errors = multierror.Append(errors, err)          continue       }       lg.Debug("Dispatching to main adapters after running preprocessors")       lg.Debuga("Attribute Bag: \n", reportBag)       lg.Debugf("Dispatching Report %d out of %d", i+1, len(req.Attributes))       // 真正开始分发,将数据逐步加入到缓存中       if err := reporter.Report(reportBag); err != nil {          span.LogFields(otlog.String("error", err.Error()))          span.Finish()          errors = multierror.Append(errors, err)          continue       }       span.Finish()       // purge the effect of the Preprocess call so that the next time through everything is clean       reportBag.Reset()    }    reportBag.Done()    accumBag.Done()    protoBag.Done()    // 真正的发送函数,从缓存中取出并发送到adaptor    if err := reporter.Flush(); err != nil {       errors = multierror.Append(errors, err)    }    reporter.Done()    if errors != nil {       reportSpan.LogFields(otlog.String("error", errors.Error()))    }    reportSpan.Finish()    if errors != nil {       lg.Errora("Report failed:", errors.Error())       return nil, grpc.Errorf(codes.Unknown, errors.Error())    }    // 过程结束    return reportResp, nil }

通过上述代码解读,我们了解了Report接口的工作流程,但此时我们还并不知道一个请求的状态是如何报给adaptor的,下面我们通过简要的函数串接,把这部分流程串起来:

上述的预处理阶段Preprocess与上报阶段Report,最终都会调用到dispatch函数,仅通过不同的type来区分要做的事情;

func (d *Impl) Preprocess(ctx context.Context, bag attribute.Bag, responseBag *attribute.MutableBag) error {    s := d.getSession(ctx, tpb.TEMPLATE_VARIETY_ATTRIBUTE_GENERATOR, bag)    s.responseBag = responseBag    err := s.dispatch()    if err == nil {       err = s.err    }    … … } func (r *reporter) Report(bag attribute.Bag) error {    s := r.impl.getSession(r.ctx, tpb.TEMPLATE_VARIETY_REPORT, bag)    s.reportStates = r.states    err := s.dispatch()    if err == nil {       err = s.err    }    … … }

而dispatch函数中做了真正的分发动作,包括:

1.遍历所有adaptor,调用adaptor中的函数,针对不同的adaptor生成不同的instance,并将instance缓存放入reportstates

var instance interface{} if instance, err = input.Builder(s.bag); err != nil {    log.Errorf("error creating instance: destination='%v', error='%v'", destination.FriendlyName, err)    s.err = multierror.Append(s.err, err)    continue } type NamedBuilder struct {    InstanceShortName string    Builder           template.InstanceBuilderFn } InstanceBuilderFn func(attrs attribute.Bag) (interface{}, error) CreateInstanceBuilder: func(instanceName string, param proto.Message, expb *compiled.ExpressionBuilder) (template.InstanceBuilderFn, error) builder.build(attr) // For report templates, accumulate instances as much as possible before commencing dispatch. if s.variety == tpb.TEMPLATE_VARIETY_REPORT {    state.instances = append(state.instances, instance)    continue }

2.将instance分发到所有adaptor,最终调用并分发到adaptor的HandleMetric函数中

func (r *reporter) Flush() error {    s := r.impl.getSession(r.ctx, tpb.TEMPLATE_VARIETY_REPORT, nil)    s.reportStates = r.states    s.dispatchBufferedReports()    err := s.err    … … } func (s *session) dispatchBufferedReports() {    // Ensure that we can run dispatches to all destinations in parallel.    s.ensureParallelism(len(s.reportStates))    // dispatch the buffered dispatchStates we've got    for k, v := range s.reportStates {       s.dispatchToHandler(v)       delete(s.reportStates, k)    }    s.waitForDispatched() } func (s *session) dispatchToHandler(ds *dispatchState) {    s.activeDispatches++    ds.session = s    s.impl.gp.ScheduleWork(ds.invokeHandler, nil) } case tpb.TEMPLATE_VARIETY_REPORT:    ds.err = ds.destination.Template.DispatchReport(       ctx, ds.destination.Handler, ds.instances) type TemplateInfo struct {    Name             string    Variety          tpb.TemplateVariety    DispatchReport   template.DispatchReportFn    DispatchCheck    template.DispatchCheckFn    DispatchQuota    template.DispatchQuotaFn    DispatchGenAttrs template.DispatchGenerateAttributesFn } DispatchReport: func(ctx context.Context, handler adapter.Handler, inst []interface{}) error {    // Convert the instances from the generic []interface{}, to their specialized type.    instances := make([]*metric.Instance, len(inst))    for i, instance := range inst {       instances[i] = instance.(*metric.Instance)    }    // Invoke the handler.    if err := handler.(metric.Handler).HandleMetric(ctx, instances); err != nil {       return fmt.Errorf("failed to report all values: %v", err)    }    return nil }

2.2 相关结构体定义

Report接口请求体定义

// Used to report telemetry after performing one or more actions. type ReportRequest struct {    // 代表一个请求中的属性    // 每个attribute代表一个请求动作,多个动作可汇总在一条message中以提高效率    //虽然每个“属性”消息在语义上被视为与消息中的其他属性无关的独立独立实体,但此消息格式利用属性消息之间的增量编码,以便大幅减少请求大小并改进端到端 效率。 每组单独的属性用于修改前一组。 这消除了在单个请求中多次冗余地发送相同属性的需要。    // 如果客户端上报时不想使用增量编码,可全量的发送所有属性.    Attributes []CompressedAttributes `protobuf:"bytes,1,rep,name=attributes" json:"attributes"`    // 所有属性的默认消息级字典.    // 这使得可以为该请求中的所有属性共享相同的字典,这可以大大减少整体请求大小    DefaultWords []string `protobuf:"bytes,2,rep,name=default_words,json=defaultWords" json:"default_words,omitempty"`    // 全局字典的词条数,可检测客户端与服务端之间的全局字典是否同步    GlobalWordCount uint32 `protobuf:"varint,3,opt,name=global_word_count,json=globalWordCount,proto3" json:"global_word_count,omitempty"` }

3、总结

Mixer中涉及很多缓存命中等用于优化性能的设计,本文仅介绍了Mixer中Report接口发送到adaptor的过程,一些性能优化设计,如protobag,dispatch缓存等内容,将会在后续文章中解析。

相关服务请访问https://support.huaweicloud.com/cce/index.html?cce_helpcenter_2019

Istio

版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。

上一篇:深度学习 GPU环境 Ubuntu 16.04 + Nvidia GTX 1080 + Python 3.6 + CUDA 9.
下一篇:专访Kohsuke Kawaguchi:如何依靠开源项目养活自己
相关文章