Merge pull request #2348 from hongliang81/develop

Add Aliyun Logger

.travis.yml

@@ -31,6 +31,8 @@ install:
   - go get github.com/siddontang/ledisdb/config
   - go get github.com/siddontang/ledisdb/ledis
   - go get github.com/ssdb/gossdb/ssdb
+  - go get github.com/cloudflare/golz4
+  - go get github.com/gogo/protobuf/proto
 before_script:
   - psql --version
   - sh -c "if [ '$ORM_DRIVER' = 'postgres' ]; then psql -c 'create database orm_test;' -U postgres; fi"

logs/alils/alils.go

+package alils
+
+import (
+	"encoding/json"
+	"github.com/astaxie/beego/logs"
+	"github.com/gogo/protobuf/proto"
+	"strings"
+	"sync"
+	"time"
+)
+
+const (
+	CacheSize int    = 64
+	Delimiter string = "##"
+)
+
+type AliLSConfig struct {
+	Project   string   `json:"project"`
+	Endpoint  string   `json:"endpoint"`
+	KeyID     string   `json:"key_id"`
+	KeySecret string   `json:"key_secret"`
+	LogStore  string   `json:"log_store"`
+	Topics    []string `json:"topics"`
+	Source    string   `json:"source"`
+	Level     int      `json:"level"`
+	FlushWhen int      `json:"flush_when"`
+}
+
+// aliLSWriter implements LoggerInterface.
+// it writes messages in keep-live tcp connection.
+type aliLSWriter struct {
+	store    *LogStore
+	group    []*LogGroup
+	withMap  bool
+	groupMap map[string]*LogGroup
+	lock     *sync.Mutex
+	AliLSConfig
+}
+
+// 创建提供Logger接口的日志服务
+func NewAliLS() logs.Logger {
+	alils := new(aliLSWriter)
+	alils.Level = logs.LevelTrace
+	return alils
+}
+
+// 读取配置
+// 初始化必要的数据结构
+func (c *aliLSWriter) Init(jsonConfig string) (err error) {
+
+	json.Unmarshal([]byte(jsonConfig), c)
+
+	if c.FlushWhen > CacheSize {
+		c.FlushWhen = CacheSize
+	}
+
+	// 初始化Project
+	prj := &LogProject{
+		Name:            c.Project,
+		Endpoint:        c.Endpoint,
+		AccessKeyId:     c.KeyID,
+		AccessKeySecret: c.KeySecret,
+	}
+
+	// 获取logstore
+	c.store, err = prj.GetLogStore(c.LogStore)
+	if err != nil {
+		return err
+	}
+
+	// 创建默认Log Group
+	c.group = append(c.group, &LogGroup{
+		Topic:  proto.String(""),
+		Source: proto.String(c.Source),
+		Logs:   make([]*Log, 0, c.FlushWhen),
+	})
+
+	// 创建其它Log Group
+	c.groupMap = make(map[string]*LogGroup)
+	for _, topic := range c.Topics {
+
+		lg := &LogGroup{
+			Topic:  proto.String(topic),
+			Source: proto.String(c.Source),
+			Logs:   make([]*Log, 0, c.FlushWhen),
+		}
+
+		c.group = append(c.group, lg)
+		c.groupMap[topic] = lg
+	}
+
+	if len(c.group) == 1 {
+		c.withMap = false
+	} else {
+		c.withMap = true
+	}
+
+	c.lock = &sync.Mutex{}
+
+	return nil
+}
+
+// WriteMsg write message in connection.
+// if connection is down, try to re-connect.
+func (c *aliLSWriter) WriteMsg(when time.Time, msg string, level int) (err error) {
+
+	if level > c.Level {
+		return nil
+	}
+
+	var topic string
+	var content string
+	var lg *LogGroup
+	if c.withMap {
+
+		// 解析出Topic，并匹配LogGroup
+		strs := strings.SplitN(msg, Delimiter, 2)
+		if len(strs) == 2 {
+			pos := strings.LastIndex(strs[0], " ")
+			topic = strs[0][pos+1 : len(strs[0])]
+			content = strs[0][0:pos] + strs[1]
+			lg = c.groupMap[topic]
+		}
+
+		// 默认发到空Topic
+		if lg == nil {
+			topic = ""
+			content = msg
+			lg = c.group[0]
+		}
+	} else {
+		topic = ""
+		content = msg
+		lg = c.group[0]
+	}
+
+	// 生成日志
+	c1 := &Log_Content{
+		Key:   proto.String("msg"),
+		Value: proto.String(content),
+	}
+
+	l := &Log{
+		Time: proto.Uint32(uint32(when.Unix())), // 填写日志时间
+		Contents: []*Log_Content{
+			c1,
+		},
+	}
+
+	c.lock.Lock()
+	lg.Logs = append(lg.Logs, l)
+	c.lock.Unlock()
+
+	// 满足条件则Flush
+	if len(lg.Logs) >= c.FlushWhen {
+		c.flush(lg)
+	}
+
+	return nil
+}
+
+// Flush implementing method. empty.
+func (c *aliLSWriter) Flush() {
+
+	// flush所有group
+	for _, lg := range c.group {
+		c.flush(lg)
+	}
+}
+
+// Destroy destroy connection writer and close tcp listener.
+func (c *aliLSWriter) Destroy() {
+}
+
+func (c *aliLSWriter) flush(lg *LogGroup) {
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	// 把以上的LogGroup推送到SLS服务器，
+	// SLS服务器会根据该logstore的shard个数自动进行负载均衡。
+	err := c.store.PutLogs(lg)
+	if err != nil {
+		return
+	}
+
+	lg.Logs = make([]*Log, 0, c.FlushWhen)
+}
+
+func init() {
+	logs.Register(logs.AdapterAliLS, NewAliLS)
+}

logs/alils/config.go

+package alils
+
+const (
+	version         = "0.5.0"     // SDK version
+	signatureMethod = "hmac-sha1" // Signature method
+
+	// OffsetNewest stands for the log head offset, i.e. the offset that will be
+	// assigned to the next message that will be produced to the shard.
+	OffsetNewest = "end"
+	// OffsetOldest stands for the oldest offset available on the logstore for a
+	// shard.
+	OffsetOldest = "begin"
+)

logs/alils/log_config.go

+package alils
+
+type InputDetail struct {
+	LogType       string   `json:"logType"`
+	LogPath       string   `json:"logPath"`
+	FilePattern   string   `json:"filePattern"`
+	LocalStorage  bool     `json:"localStorage"`
+	TimeFormat    string   `json:"timeFormat"`
+	LogBeginRegex string   `json:"logBeginRegex"`
+	Regex         string   `json:"regex"`
+	Keys          []string `json:"key"`
+	FilterKeys    []string `json:"filterKey"`
+	FilterRegex   []string `json:"filterRegex"`
+	TopicFormat   string   `json:"topicFormat"`
+}
+
+type OutputDetail struct {
+	Endpoint     string `json:"endpoint"`
+	LogStoreName string `json:"logstoreName"`
+}
+
+type LogConfig struct {
+	Name         string       `json:"configName"`
+	InputType    string       `json:"inputType"`
+	InputDetail  InputDetail  `json:"inputDetail"`
+	OutputType   string       `json:"outputType"`
+	OutputDetail OutputDetail `json:"outputDetail"`
+
+	CreateTime     uint32
+	LastModifyTime uint32
+
+	project *LogProject
+}
+
+// GetAppliedMachineGroup returns applied machine group of this config.
+func (c *LogConfig) GetAppliedMachineGroup(confName string) (groupNames []string, err error) {
+	groupNames, err = c.project.GetAppliedMachineGroups(c.Name)
+	return
+}

logs/alils/log_store.go

+package alils
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/http/httputil"
+	"strconv"
+
+	lz4 "github.com/cloudflare/golz4"
+	"github.com/gogo/protobuf/proto"
+)
+
+type LogStore struct {
+	Name       string `json:"logstoreName"`
+	TTL        int
+	ShardCount int
+
+	CreateTime     uint32
+	LastModifyTime uint32
+
+	project *LogProject
+}
+
+type Shard struct {
+	ShardID int `json:"shardID"`
+}
+
+// ListShards returns shard id list of this logstore.
+func (s *LogStore) ListShards() (shardIDs []int, err error) {
+	h := map[string]string{
+		"x-sls-bodyrawsize": "0",
+	}
+
+	uri := fmt.Sprintf("/logstores/%v/shards", s.Name)
+	r, err := request(s.project, "GET", uri, h, nil)
+	if err != nil {
+		return
+	}
+
+	buf, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return
+	}
+
+	if r.StatusCode != http.StatusOK {
+		errMsg := &errorMessage{}
+		err = json.Unmarshal(buf, errMsg)
+		if err != nil {
+			err = fmt.Errorf("failed to list logstore")
+			dump, _ := httputil.DumpResponse(r, true)
+			fmt.Println(dump)
+			return
+		}
+		err = fmt.Errorf("%v:%v", errMsg.Code, errMsg.Message)
+		return
+	}
+
+	var shards []*Shard
+	err = json.Unmarshal(buf, &shards)
+	if err != nil {
+		return
+	}
+
+	for _, v := range shards {
+		shardIDs = append(shardIDs, v.ShardID)
+	}
+	return
+}
+
+// PutLogs put logs into logstore.
+// The callers should transform user logs into LogGroup.
+func (s *LogStore) PutLogs(lg *LogGroup) (err error) {
+	body, err := proto.Marshal(lg)
+	if err != nil {
+		return
+	}
+
+	// Compresse body with lz4
+	out := make([]byte, lz4.CompressBound(body))
+	n, err := lz4.Compress(body, out)
+	if err != nil {
+		return
+	}
+
+	h := map[string]string{
+		"x-sls-compresstype": "lz4",
+		"x-sls-bodyrawsize":  fmt.Sprintf("%v", len(body)),
+		"Content-Type":       "application/x-protobuf",
+	}
+
+	uri := fmt.Sprintf("/logstores/%v", s.Name)
+	r, err := request(s.project, "POST", uri, h, out[:n])
+	if err != nil {
+		return
+	}
+
+	buf, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return
+	}
+
+	if r.StatusCode != http.StatusOK {
+		errMsg := &errorMessage{}
+		err = json.Unmarshal(buf, errMsg)
+		if err != nil {
+			err = fmt.Errorf("failed to put logs")
+			dump, _ := httputil.DumpResponse(r, true)
+			fmt.Println(dump)
+			return
+		}
+		err = fmt.Errorf("%v:%v", errMsg.Code, errMsg.Message)
+		return
+	}
+	return
+}
+
+// GetCursor gets log cursor of one shard specified by shardId.
+// The from can be in three form: a) unix timestamp in seccond, b) "begin", c) "end".
+// For more detail please read: http://gitlab.alibaba-inc.com/sls/doc/blob/master/api/shard.md#logstore
+func (s *LogStore) GetCursor(shardId int, from string) (cursor string, err error) {
+	h := map[string]string{
+		"x-sls-bodyrawsize": "0",
+	}
+
+	uri := fmt.Sprintf("/logstores/%v/shards/%v?type=cursor&from=%v",
+		s.Name, shardId, from)
+
+	r, err := request(s.project, "GET", uri, h, nil)
+	if err != nil {
+		return
+	}
+
+	buf, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return
+	}
+
+	if r.StatusCode != http.StatusOK {
+		errMsg := &errorMessage{}
+		err = json.Unmarshal(buf, errMsg)
+		if err != nil {
+			err = fmt.Errorf("failed to get cursor")
+			dump, _ := httputil.DumpResponse(r, true)
+			fmt.Println(dump)
+			return
+		}
+		err = fmt.Errorf("%v:%v", errMsg.Code, errMsg.Message)
+		return
+	}
+
+	type Body struct {
+		Cursor string
+	}
+	body := &Body{}
+
+	err = json.Unmarshal(buf, body)
+	if err != nil {
+		return
+	}
+	cursor = body.Cursor
+	return
+}
+
+// GetLogsBytes gets logs binary data from shard specified by shardId according cursor.
+// The logGroupMaxCount is the max number of logGroup could be returned.
+// The nextCursor is the next curosr can be used to read logs at next time.
+func (s *LogStore) GetLogsBytes(shardId int, cursor string,
+	logGroupMaxCount int) (out []byte, nextCursor string, err error) {
+
+	h := map[string]string{
+		"x-sls-bodyrawsize": "0",
+		"Accept":            "application/x-protobuf",
+		"Accept-Encoding":   "lz4",
+	}
+
+	uri := fmt.Sprintf("/logstores/%v/shards/%v?type=logs&cursor=%v&count=%v",
+		s.Name, shardId, cursor, logGroupMaxCount)
+
+	r, err := request(s.project, "GET", uri, h, nil)
+	if err != nil {
+		return
+	}
+
+	buf, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return
+	}
+
+	if r.StatusCode != http.StatusOK {
+		errMsg := &errorMessage{}
+		err = json.Unmarshal(buf, errMsg)
+		if err != nil {
+			err = fmt.Errorf("failed to get cursor")
+			dump, _ := httputil.DumpResponse(r, true)
+			fmt.Println(dump)
+			return
+		}
+		err = fmt.Errorf("%v:%v", errMsg.Code, errMsg.Message)
+		return
+	}
+
+	v, ok := r.Header["X-Sls-Compresstype"]
+	if !ok || len(v) == 0 {
+		err = fmt.Errorf("can't find 'x-sls-compresstype' header")
+		return
+	}
+	if v[0] != "lz4" {
+		err = fmt.Errorf("unexpected compress type:%v", v[0])
+		return
+	}
+
+	v, ok = r.Header["X-Sls-Cursor"]
+	if !ok || len(v) == 0 {
+		err = fmt.Errorf("can't find 'x-sls-cursor' header")
+		return
+	}
+	nextCursor = v[0]
+
+	v, ok = r.Header["X-Sls-Bodyrawsize"]
+	if !ok || len(v) == 0 {
+		err = fmt.Errorf("can't find 'x-sls-bodyrawsize' header")
+		return
+	}
+	bodyRawSize, err := strconv.Atoi(v[0])
+	if err != nil {
+		return
+	}
+
+	out = make([]byte, bodyRawSize)
+	err = lz4.Uncompress(buf, out)
+	if err != nil {
+		return
+	}
+
+	return
+}
+
+// LogsBytesDecode decodes logs binary data retruned by GetLogsBytes API
+func LogsBytesDecode(data []byte) (gl *LogGroupList, err error) {
+
+	gl = &LogGroupList{}
+	err = proto.Unmarshal(data, gl)
+	if err != nil {
+		return
+	}
+
+	return
+}
+
+// GetLogs gets logs from shard specified by shardId according cursor.
+// The logGroupMaxCount is the max number of logGroup could be returned.
+// The nextCursor is the next curosr can be used to read logs at next time.
+func (s *LogStore) GetLogs(shardId int, cursor string,
+	logGroupMaxCount int) (gl *LogGroupList, nextCursor string, err error) {
+
+	out, nextCursor, err := s.GetLogsBytes(shardId, cursor, logGroupMaxCount)
+	if err != nil {
+		return
+	}
+
+	gl, err = LogsBytesDecode(out)
+	if err != nil {
+		return
+	}
+
+	return
+}

logs/alils/machine_group.go

+package alils
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/http/httputil"
+)
+
+type MachinGroupAttribute struct {
+	ExternalName string `json:"externalName"`
+	TopicName    string `json:"groupTopic"`
+}
+
+type MachineGroup struct {
+	Name          string   `json:"groupName"`
+	Type          string   `json:"groupType"`
+	MachineIdType string   `json:"machineIdentifyType"`
+	MachineIdList []string `json:"machineList"`
+
+	Attribute MachinGroupAttribute `json:"groupAttribute"`
+
+	CreateTime     uint32
+	LastModifyTime uint32
+
+	project *LogProject
+}
+
+type Machine struct {
+	IP            string
+	UniqueId      string `json:"machine-uniqueid"`
+	UserdefinedId string `json:"userdefined-id"`
+}
+
+type MachineList struct {
+	Total    int
+	Machines []*Machine
+}
+
+// ListMachines returns machine list of this machine group.
+func (m *MachineGroup) ListMachines() (ms []*Machine, total int, err error) {
+	h := map[string]string{
+		"x-sls-bodyrawsize": "0",
+	}
+
+	uri := fmt.Sprintf("/machinegroups/%v/machines", m.Name)
+	r, err := request(m.project, "GET", uri, h, nil)
+	if err != nil {
+		return
+	}
+
+	buf, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return
+	}
+
+	if r.StatusCode != http.StatusOK {
+		errMsg := &errorMessage{}
+		err = json.Unmarshal(buf, errMsg)
+		if err != nil {
+			err = fmt.Errorf("failed to remove config from machine group")
+			dump, _ := httputil.DumpResponse(r, true)
+			fmt.Println(dump)
+			return
+		}
+		err = fmt.Errorf("%v:%v", errMsg.Code, errMsg.Message)
+		return
+	}
+
+	body := &MachineList{}
+	err = json.Unmarshal(buf, body)
+	if err != nil {
+		return
+	}
+
+	ms = body.Machines
+	total = body.Total
+
+	return
+}
+
+// GetAppliedConfigs returns applied configs of this machine group.
+func (m *MachineGroup) GetAppliedConfigs() (confNames []string, err error) {
+	confNames, err = m.project.GetAppliedConfigs(m.Name)
+	return
+}

logs/alils/request.go

+package alils
+
+import (
+	"bytes"
+	"crypto/md5"
+	"fmt"
+	"net/http"
+)
+
+// request sends a request to SLS.
+func request(project *LogProject, method, uri string, headers map[string]string,
+	body []byte) (resp *http.Response, err error) {
+
+	// The caller should provide 'x-sls-bodyrawsize' header
+	if _, ok := headers["x-sls-bodyrawsize"]; !ok {
+		err = fmt.Errorf("Can't find 'x-sls-bodyrawsize' header")
+		return
+	}
+
+	// SLS public request headers
+	headers["Host"] = project.Name + "." + project.Endpoint
+	headers["Date"] = nowRFC1123()
+	headers["x-sls-apiversion"] = version
+	headers["x-sls-signaturemethod"] = signatureMethod
+	if body != nil {
+		bodyMD5 := fmt.Sprintf("%X", md5.Sum(body))
+		headers["Content-MD5"] = bodyMD5
+
+		if _, ok := headers["Content-Type"]; !ok {
+			err = fmt.Errorf("Can't find 'Content-Type' header")
+			return
+		}
+	}
+
+	// Calc Authorization
+	// Authorization = "SLS <AccessKeyId>:<Signature>"
+	digest, err := signature(project, method, uri, headers)
+	if err != nil {
+		return
+	}
+	auth := fmt.Sprintf("SLS %v:%v", project.AccessKeyId, digest)
+	headers["Authorization"] = auth
+
+	// Initialize http request
+	reader := bytes.NewReader(body)
+	urlStr := fmt.Sprintf("http://%v.%v%v", project.Name, project.Endpoint, uri)
+	req, err := http.NewRequest(method, urlStr, reader)
+	if err != nil {
+		return
+	}
+	for k, v := range headers {
+		req.Header.Add(k, v)
+	}
+
+	// Get ready to do request
+	resp, err = http.DefaultClient.Do(req)
+	if err != nil {
+		return
+	}
+
+	return
+}

logs/alils/signature.go

+package alils
+
+import (
+	"crypto/hmac"
+	"crypto/sha1"
+	"encoding/base64"
+	"fmt"
+	"net/url"
+	"sort"
+	"strings"
+	"time"
+)
+
+// GMT location
+var gmtLoc = time.FixedZone("GMT", 0)
+
+// NowRFC1123 returns now time in RFC1123 format with GMT timezone,
+// eg. "Mon, 02 Jan 2006 15:04:05 GMT".
+func nowRFC1123() string {
+	return time.Now().In(gmtLoc).Format(time.RFC1123)
+}
+
+// signature calculates a request's signature digest.
+func signature(project *LogProject, method, uri string,
+	headers map[string]string) (digest string, err error) {
+	var contentMD5, contentType, date, canoHeaders, canoResource string
+	var slsHeaderKeys sort.StringSlice
+
+	// SignString = VERB + "\n"
+	//              + CONTENT-MD5 + "\n"
+	//              + CONTENT-TYPE + "\n"
+	//              + DATE + "\n"
+	//              + CanonicalizedSLSHeaders + "\n"
+	//              + CanonicalizedResource
+
+	if val, ok := headers["Content-MD5"]; ok {
+		contentMD5 = val
+	}
+
+	if val, ok := headers["Content-Type"]; ok {
+		contentType = val
+	}
+
+	date, ok := headers["Date"]
+	if !ok {
+		err = fmt.Errorf("Can't find 'Date' header")
+		return
+	}
+
+	// Calc CanonicalizedSLSHeaders
+	slsHeaders := make(map[string]string, len(headers))
+	for k, v := range headers {
+		l := strings.TrimSpace(strings.ToLower(k))
+		if strings.HasPrefix(l, "x-sls-") {
+			slsHeaders[l] = strings.TrimSpace(v)
+			slsHeaderKeys = append(slsHeaderKeys, l)
+		}
+	}
+
+	sort.Sort(slsHeaderKeys)
+	for i, k := range slsHeaderKeys {
+		canoHeaders += k + ":" + slsHeaders[k]
+		if i+1 < len(slsHeaderKeys) {
+			canoHeaders += "\n"
+		}
+	}
+
+	// Calc CanonicalizedResource
+	u, err := url.Parse(uri)
+	if err != nil {
+		return
+	}
+
+	canoResource += url.QueryEscape(u.Path)
+	if u.RawQuery != "" {
+		var keys sort.StringSlice
+
+		vals := u.Query()
+		for k, _ := range vals {
+			keys = append(keys, k)
+		}
+
+		sort.Sort(keys)
+		canoResource += "?"
+		for i, k := range keys {
+			if i > 0 {
+				canoResource += "&"
+			}
+
+			for _, v := range vals[k] {
+				canoResource += k + "=" + v
+			}
+		}
+	}
+
+	signStr := method + "\n" +
+		contentMD5 + "\n" +
+		contentType + "\n" +
+		date + "\n" +
+		canoHeaders + "\n" +
+		canoResource
+
+	// Signature = base64(hmac-sha1(UTF8-Encoding-Of(SignString)，AccessKeySecret))
+	mac := hmac.New(sha1.New, []byte(project.AccessKeySecret))
+	_, err = mac.Write([]byte(signStr))
+	if err != nil {
+		return
+	}
+	digest = base64.StdEncoding.EncodeToString(mac.Sum(nil))
+	return
+}
+

logs/log.go

@@ -71,6 +71,7 @@ const (
 	AdapterEs        = "es"
 	AdapterJianLiao  = "jianliao"
 	AdapterSlack     = "slack"
+	AdapterAliLS     = "alils"
 )
 
 // Legacy log level constants to ensure backwards compatibility.