http flusher 支持自定义 authenticator 方案

[snakorse]

需求

• http flusher 需要支持请求加签、追加auth header，请求的加签算法可能因后端平台而已，因此，需要能够通过某种形式支持自定义 authenticator 实现

方案

• 定义 ClientAuthenticator 接口

// ClientAuthenticator allow adding auth or signature info to HTTP client request
type ClientAuthenticator interface {
   // RoundTripper returns a RoundTripper that can be used to authenticate HTTP requests
   RoundTripper(base http.RoundTripper) (http.RoundTripper, error)
}

• Pipeline 定义新的插件类型 Extension

// Extension ...
type Extension interface {
   // Description returns a one-sentence description on the Extension
   Description() string

   // Init called for init some system resources, like socket, mutex...
   Init(Context) error

   // Stop stops the services and release resources
   Stop() error
}

type ExtensionV1 interface {
   Extension
}

type ExtensionV2 interface {
   Extension
}

• 自定义的 ClientAuthenticator 实现，通过 pipeline config 的 extensions 配置节注册

{
    "flushers": [
        {
            "type": "flusher_http",
            "detail": {
                "authenticator": "extension_basicauth/ba_for_user1"    //extensiontype/id
            }
        }
    ],
    "extensions": [
        {
            "type": "extension_basicauth/ba_for_user1",
            "detail": {
                "Username": "user1",
                "Password": "pwd1"
            }
        }
    ]
}

注： 由于在同一pipeline中，可能存在同一类型的extension的多个实例，因此type做了扩展：type/id, 增加了id部分用来区分同一类型的多个extension

• pipeline.Context 增加方法 GetExtension(name string) (Extension,bool)，以支持在其他插件中获取extension实例使用

// Context for plugin
type Context interface {
   InitContext(project, logstore, configName string)

   GetConfigName() string
   GetProject() string
   GetLogstore() string
   GetRuntimeContext() context.Context
   GetExtension(name string) (Extension, bool)
}


func (f *FlusherHTTP) initHTTPClient() error {
   transport := http.DefaultTransport
   var err error
   if dt, ok := transport.(*http.Transport); ok {
      dt = dt.Clone()
      if f.Concurrency > dt.MaxIdleConnsPerHost {
         dt.MaxIdleConnsPerHost = f.Concurrency + 1
      }
      transport = dt
   }
   if f.Authenticator != "" {
      auth, ok := f.context.GetExtension(f.Authenticator)
      if !ok {
         err = fmt.Errorf("not found authenticator[%s]", f.Authenticator)
         logger.Error(f.context.GetRuntimeContext(), "FLUSHER_INIT_ALARM", "http flusher init authenticator fail, error", err)
         return err
      }
      ca, ok := auth.(extensions.ClientAuthenticator)
      if !ok {
         err = fmt.Errorf("authenticator(%s) not implement interface extensions.ClientAuthenticator", f.Authenticator)
         logger.Error(f.context.GetRuntimeContext(), "FLUSHER_INIT_ALARM", "http flusher init authenticator fail, error", err)
         return err
      }
      transport, err = ca.RoundTripper(transport)
      if err != nil {
         logger.Error(f.context.GetRuntimeContext(), "FLUSHER_INIT_ALARM", "http flusher init authenticator fail, error", err)
         return err
      }
   }

   f.client = &http.Client{
      Timeout:   f.Timeout,
      Transport: transport,
   }
   return nil
}

[snakorse]

pull request: #648

[liuhaoyang]

有几个疑问

• extension 代码依赖的方式，是不是需要所有的extension 接口都要定义到 pkg 包里
• extension v1和v2的区别是什么，如果定义相同的话是不是能不区分？
• 看上去你的需求是做一些接口的静态逻辑实现，那如果要读取远程配置的数据到插件里，怎么用这个机制实现？
• extension 已知的类型有哪些？

[liuhaoyang]

Extension 插件方案需要加一些更详细的描述，对有状态和无状态的extension的处理估计是不一样的