deploy/all/prometheus/main.go

package main

import (
	"log"
	"net/http"
	"os"
	"strconv"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"

	sls "github.com/aliyun/aliyun-log-go-sdk"
)

var (
	AccessKeyID              = "LTAIWYUujJAm7CbH"
	AccessKeySecret          = "RfSjdiWwED1sGFlsjXv0DlfTnZTG1P"
	Endpoint                 = "cn-hangzhou.log.aliyuncs.com"
	adActionLogStore         *sls.LogStore
	longvideoReqeustLogStore *sls.LogStore
	videoPlayStore           *sls.LogStore
	videoActionStore         *sls.LogStore
	simpleeventStore         *sls.LogStore
	userShareStore           *sls.LogStore
	adRequestStore           *sls.LogStore

	adReportQuerySQL  = `* | SELECT appType, json_extract_scalar(extParams, '$.eventInfos.ab_test002') AS adGroup, CASE WHEN adType = 1 THEN 'BANNER广告' WHEN adType = 2 THEN '视频广告' WHEN adType = 3 THEN '贴片广告' WHEN adType = 4 THEN '插屏广告' WHEN adType = 5 THEN '开屏广告' WHEN adType = 6 THEN '激励广告' ELSE ' ' END AS adType, COUNT( DISTINCT CASE WHEN businessType = 'adLoaded' THEN machineCode END ) AS adLoaded, COUNT( DISTINCT CASE WHEN businessType = 'adRequest' THEN machineCode END ) AS adRequest, COUNT( DISTINCT CASE WHEN businessType = 'adView' THEN machineCode END ) AS adView, COUNT( DISTINCT CASE WHEN businessType = 'adClick' THEN machineCode END ) AS adClick, COUNT( DISTINCT CASE WHEN businessType = 'adShowError' THEN machineCode END ) AS adShowError, COUNT( DISTINCT CASE WHEN businessType = 'adLoadError' THEN machineCode END ) AS adLoadError, COUNT( DISTINCT CASE WHEN businessType = 'adConfigRequest' THEN machineCode END ) AS adConfigRequest, COUNT( DISTINCT CASE WHEN businessType = 'adConfigLoaded' THEN machineCode END ) AS adConfigLoaded GROUP BY adGroup, adType, appType HAVING adGroup IS NOT NULL AND adType != ' ' ORDER BY appType ASC LIMIT 1000`
	adReportQuerySQL4 = `* | SELECT appType, json_extract_scalar(extParams, '$.eventInfos.ab_test004') AS adGroup, CASE WHEN adType = 1 THEN 'BANNER广告' WHEN adType = 2 THEN '视频广告' WHEN adType = 3 THEN '贴片广告' WHEN adType = 4 THEN '插屏广告' WHEN adType = 5 THEN '开屏广告' WHEN adType = 6 THEN '激励广告' ELSE ' ' END AS adType, COUNT( DISTINCT CASE WHEN businessType = 'adLoaded' THEN machineCode END ) AS adLoaded, COUNT( DISTINCT CASE WHEN businessType = 'adRequest' THEN machineCode END ) AS adRequest, COUNT( DISTINCT CASE WHEN businessType = 'adView' THEN machineCode END ) AS adView, COUNT( DISTINCT CASE WHEN businessType = 'adClick' THEN machineCode END ) AS adClick, COUNT( DISTINCT CASE WHEN businessType = 'adShowError' THEN machineCode END ) AS adShowError, COUNT( DISTINCT CASE WHEN businessType = 'adLoadError' THEN machineCode END ) AS adLoadError, COUNT( DISTINCT CASE WHEN businessType = 'adConfigRequest' THEN machineCode END ) AS adConfigRequest, COUNT( DISTINCT CASE WHEN businessType = 'adConfigLoaded' THEN machineCode END ) AS adConfigLoaded GROUP BY adGroup, adType, appType HAVING adGroup IS NOT NULL AND adType != ' ' ORDER BY appType ASC LIMIT 1000`
	apiQuerySQL       = `* and requestUri: "/longvideoapi/video/distribute/category/videoList/v2" or requestUri: "/longvideoapi/video/v2/detail" or requestUri: "/longvideoapi/video/v2/sharePageDetail" or requestUri: "/longvideoapi/video/recommend/sharePage/list" | SELECT requestUri, json_extract_scalar(params, '$.appType') as appType, COUNT(*) as total, count( case when resultCode = 0 THEN 1 else null end ) as success, count( case when resultCode != 0 THEN 1 else null end ) as error GROUP by appType, requestUri order by requestUri asc limit 1000`
	videoPlayQuerySQL = `* | select appType, COUNT(*) as cnt, businessType, pageSource GROUP by appType, businessType, pageSource ORDER BY appType ASC, cnt DESC LIMIT 1000`
	// videoPlayCountQuerySQL = `* AND businessType : videoPlay | select appType, COUNT(*) as plyaCount, pageSource GROUP BY appType, pageSource ORDER BY appType ASC`

	// adConfigQuerySQL = `* and requestUri: /longvideoapi/ad/position/info | select json_extract_scalar(params, '$.appType') as appType,count(*) as total, count(case when resultCode = 0 then 1 else null end) as success,count(case when resultCode != 0 then 1 else null end) as error,count(case when resultSize > 0 then 1 else null end) as hasAds,count(case when resultSize <= 0 then 1 else null end) as noAds group by appType`

	adConfigQuerySQL = `* and url : "/ad/position/info"  | SELECT count(*) as total, count( case when json_extract_scalar(responseBody, '$.code') = '0' then 1 else null end ) as success, count( case when json_extract_scalar(responseBody, '$.code') != '0' then 1 else null end ) as error, count( case when json_array_length(json_extract(responseBody, '$.data')) <= 0 then 1 else null end ) as noAds, count( case when json_array_length(json_extract(responseBody, '$.data')) > 0 then 1 else null end ) as hasAds, json_extract(requestBody, '$.baseInfo.appType') as appType group by appType `

	apiRtQuerySQL           = `* and requestUri: "/longvideoapi/video/distribute/category/videoList/v2" or requestUri: "/longvideoapi/video/v2/detail" or requestUri: "/longvideoapi/video/v2/sharePageDetail" or requestUri: "/longvideoapi/video/recommend/sharePage/list" | select "requestUri", "params.appType" as "appType" , avg("expendTime") as "rt" from log group by "requestUri", "appType" order by "appType" asc limit 1000`
	videoDataQuerySQL       = `* | select appType, COUNT(*) as cnt, businessType, pageSource GROUP by appType, businessType, pageSource ORDER BY appType ASC, businessType LIMIT 1000`
	distributeApiQuerySQL   = ` * and (requestUri: "/longvideoapi/video/distribute/category/videoList/v2" or requestUri: "/longvideoapi/video/recommend/sharePage/list") and not resultSize: 4 | SELECT json_extract_scalar(params, '$.appType') as appType, requestUri, COUNT( DISTINCT CASE WHEN resultSize = 0 THEN json_extract_scalar(params, '$.machineCode') END ) AS "等于0", COUNT( DISTINCT CASE WHEN (resultSize < 4 and resultSize >0) THEN json_extract_scalar(params, '$.machineCode') END ) AS "小于4" GROUP BY requestUri, appType  ORDER BY appType ASC LIMIT 1000`
	shareDataQuerySQL       = `* and ( businessType = 'videoShareFriend' OR businessType = 'videoShareH5' OR businessType = 'videoPublish' ) | select appType, COUNT(DISTINCT mid) as cnt,  COUNT(*) as cnt1, businessType, pageSource GROUP by appType, pageSource, businessType ORDER BY appType ASC, cnt DESC LIMIT 1000 `
	shareOperation1QuerySQL = `* and objectType: weapp_quitbutton OR objectType: weapp_share_navigator_moveVideo OR objectType: weapp_share_upload OR ( objectType: weapp_share_shareFriendWindow AND businessType: buttonClick ) | select objectType, appType, COUNT(*) as cnt GROUP by objectType, appType ORDER BY appType asc, objectType asc `
	shareOperation2QuerySQL = `* and pageSource: "user-videos-share" | select COUNT(*) as cnt, ( CASE WHEN groupShare IS NULL THEN 'shareFriend' WHEN groupShare = 1 THEN 'shareGroup' ELSE ' ' END ) as type, appType GROUP BY type, appType ORDER BY appType asc, type asc `
)

func init() {
	client := sls.CreateNormalInterface(Endpoint, AccessKeyID, AccessKeySecret, "")
	var err error

	longvideoReqeustLogStore, err = client.GetLogStore("wqsd-longvideoapi", "request-log")
	if err != nil {
		panic(err)
	}

	adActionLogStore, err = client.GetLogStore("wqsd-longvideo-frontend-log-prd", "ad-action-log")
	if err != nil {
		panic(err)
	}

	videoActionStore, err = client.GetLogStore("wqsd-video", "video-action-log")
	if err != nil {
		panic(err)
	}

	videoPlayStore, err = client.GetLogStore("wqsd-video", "video-play-log")
	if err != nil {
		panic(err)
	}

	simpleeventStore, err = client.GetLogStore("wqsd-longvideo-frontend-log-prd", "simpleevent-log")
	if err != nil {
		panic(err)
	}

	userShareStore, err = client.GetLogStore("wqsd-user", "user-share-log")
	if err != nil {
		panic(err)
	}

	adRequestStore, err = client.GetLogStore("ms-ad", "request-log")
	if err != nil {
		panic(err)
	}
}

func (c *CustomCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- c.AdInfoDesc
	ch <- c.ApiInfoDesc
	ch <- c.VideoPlayDesc
	ch <- c.AdConfigDesc
	ch <- c.AdInfo4Desc
	ch <- c.ApiRtDesc
	ch <- c.VideoDataDesc
	ch <- c.DistributeApiDesc
	ch <- c.ShareDataDesc
	ch <- c.ShareOperationDesc
}

func (c *CustomCollector) Collect(ch chan<- prometheus.Metric) {
	resp, err := longvideoReqeustLogStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), apiQuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get api logs error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		for k, v2 := range v {
			if k == "total" ||
				k == "success" ||
				k == "error" {
				v3, _ := strconv.ParseFloat(v2, 64)
				ch <- prometheus.MustNewConstMetric(
					c.ApiInfoDesc,
					prometheus.GaugeValue,
					v3,
					v["appType"], v["requestUri"], k,
				)
			}
		}
	}

	resp, err = adActionLogStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), adReportQuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get ad report logs ab_test002 error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		for k, v2 := range v {
			if k == "adRequest" ||
				k == "adShowError" ||
				k == "adLoadError" ||
				k == "adConfigRequest" ||
				k == "adConfigLoaded" ||
				k == "adLoaded" ||
				k == "adView" ||
				k == "adClick" {
				v3, _ := strconv.ParseFloat(v2, 64)
				ch <- prometheus.MustNewConstMetric(
					c.AdInfoDesc,
					prometheus.GaugeValue,
					v3,
					v["appType"], v["adGroup"], v["adType"], k,
				)
			}
		}
	}
	resp, err = adActionLogStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), adReportQuerySQL4, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get ad report logs ab_test004 error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		for k, v2 := range v {
			if k == "adRequest" ||
				k == "adShowError" ||
				k == "adLoadError" ||
				k == "adConfigRequest" ||
				k == "adConfigLoaded" ||
				k == "adLoaded" ||
				k == "adView" ||
				k == "adClick" {
				v3, _ := strconv.ParseFloat(v2, 64)
				ch <- prometheus.MustNewConstMetric(
					c.AdInfo4Desc,
					prometheus.GaugeValue,
					v3,
					v["appType"], v["adGroup"], v["adType"], k,
				)
			}
		}
	}

	resp, err = adRequestStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), adConfigQuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get ad config logs error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		for k, v2 := range v {
			if k == "total" ||
				k == "success" ||
				k == "error" ||
				k == "hasAds" ||
				k == "noAds" {
				v3, _ := strconv.ParseFloat(v2, 64)
				ch <- prometheus.MustNewConstMetric(
					c.AdConfigDesc,
					prometheus.GaugeValue,
					v3,
					v["appType"], k,
				)
			}
		}
	}

	// resp, err = videoActionStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), videoPlayCountQuerySQL, 0, 0, true)
	// if err != nil || resp == nil {
	// 	log.Printf("get video plyaCount logs error %+v resp is %+v.", err, resp)
	// 	return
	// }
	// for _, v := range resp.Logs {
	// 	v2, _ := strconv.ParseFloat(v["plyaCount"], 64)
	// 	ch <- prometheus.MustNewConstMetric(
	// 		c.VideoPlayDesc,
	// 		prometheus.GaugeValue,
	// 		v2,
	// 		v["appType"], "videoPlayTotal", v["pageSource"],
	// 	)
	// }

	resp, err = videoPlayStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), videoPlayQuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get video play logs error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		v2, _ := strconv.ParseFloat(v["cnt"], 64)
		ch <- prometheus.MustNewConstMetric(
			c.VideoPlayDesc,
			prometheus.GaugeValue,
			v2,
			v["appType"], v["businessType"], v["pageSource"],
		)
	}

	resp, err = longvideoReqeustLogStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), apiRtQuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get api rt logs error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		v3, _ := strconv.ParseFloat(v["rt"], 64)
		ch <- prometheus.MustNewConstMetric(
			c.ApiRtDesc,
			prometheus.GaugeValue,
			v3,
			v["appType"], v["requestUri"],
		)
	}

	resp, err = videoActionStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), videoDataQuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get video data logs error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		v3, _ := strconv.ParseFloat(v["cnt"], 64)
		ch <- prometheus.MustNewConstMetric(
			c.VideoDataDesc,
			prometheus.GaugeValue,
			v3,
			v["appType"], v["businessType"], v["pageSource"],
		)
	}

	resp, err = longvideoReqeustLogStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), distributeApiQuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get distribute api logs error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		// log.Printf("get distribute api logs  %+v \n", v)
		for k, v2 := range v {
			if k == "等于0" ||
				k == "小于4" {
				v3, _ := strconv.ParseFloat(v2, 64)
				ch <- prometheus.MustNewConstMetric(
					c.DistributeApiDesc,
					prometheus.GaugeValue,
					v3,
					v["appType"], v["requestUri"], k,
				)
			}
		}
	}

	resp, err = videoActionStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), shareDataQuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get share data logs error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		v3, _ := strconv.ParseFloat(v["cnt"], 64)
		ch <- prometheus.MustNewConstMetric(
			c.ShareDataDesc,
			prometheus.GaugeValue,
			v3,
			v["appType"], v["businessType"], v["pageSource"], "user",
		)

		v3, _ = strconv.ParseFloat(v["cnt1"], 64)
		ch <- prometheus.MustNewConstMetric(
			c.ShareDataDesc,
			prometheus.GaugeValue,
			v3,
			v["appType"], v["businessType"], v["pageSource"], "all",
		)
	}

	resp, err = simpleeventStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), shareOperation1QuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get share operation1 logs error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		v3, _ := strconv.ParseFloat(v["cnt"], 64)
		ch <- prometheus.MustNewConstMetric(
			c.ShareOperationDesc,
			prometheus.GaugeValue,
			v3,
			v["appType"], v["objectType"],
		)
	}
	resp, err = userShareStore.GetLogs("", time.Now().Unix()-60, time.Now().Unix(), shareOperation2QuerySQL, 0, 0, true)
	if err != nil || resp == nil {
		log.Printf("get share operation2 logs error %+v resp is %+v.", err, resp)
		return
	}
	for _, v := range resp.Logs {
		v3, _ := strconv.ParseFloat(v["cnt"], 64)
		ch <- prometheus.MustNewConstMetric(
			c.ShareOperationDesc,
			prometheus.GaugeValue,
			v3,
			v["appType"], v["type"],
		)
	}
}

type CustomCollector struct {
	AdInfoDesc   *prometheus.Desc
	ApiInfoDesc  *prometheus.Desc
	AdConfigDesc *prometheus.Desc
	AdInfo4Desc  *prometheus.Desc
	// 视频播放
	VideoPlayDesc *prometheus.Desc
	// 接口响应时间
	ApiRtDesc *prometheus.Desc
	// 播放数据
	VideoDataDesc *prometheus.Desc
	// 分发服务
	DistributeApiDesc *prometheus.Desc
	// 分享卡片数据
	ShareDataDesc *prometheus.Desc
	// 分享卡片操作
	ShareOperationDesc *prometheus.Desc
}

func main() {
	// 注册一个采集器
	reg := prometheus.NewPedanticRegistry()
	reg.MustRegister(&CustomCollector{
		AdInfoDesc: prometheus.NewDesc(
			"ad_metrics_info",
			"Front report ad metrics info",
			[]string{"app_type", "ad_group", "ad_type", "_type"},
			prometheus.Labels{},
		),
		ApiInfoDesc: prometheus.NewDesc(
			"api_metrics_info",
			"Core API Metrics Info",
			[]string{"app_type", "url", "_type"},
			prometheus.Labels{},
		),
		VideoPlayDesc: prometheus.NewDesc(
			"video_play_metrics_info",
			"Video Play Metrics Info",
			[]string{"app_type", "_type", "source"},
			prometheus.Labels{},
		),
		AdConfigDesc: prometheus.NewDesc(
			"ad_config_metrics_info",
			"Ad system deal request metrics info",
			[]string{"app_type", "_type"},
			prometheus.Labels{},
		),
		AdInfo4Desc: prometheus.NewDesc(
			"ad_metrics_info_004",
			"Front report ad metrics info ab_test004",
			[]string{"app_type", "ad_group", "ad_type", "_type"},
			prometheus.Labels{},
		),
		ApiRtDesc: prometheus.NewDesc(
			"api_metrics_rt",
			"Core API Metrics Response Time",
			[]string{"app_type", "url"},
			prometheus.Labels{},
		),
		VideoDataDesc: prometheus.NewDesc(
			"video_data_metrics",
			"Video Data Metrics",
			[]string{"app_type", "_type", "source"},
			prometheus.Labels{},
		),
		DistributeApiDesc: prometheus.NewDesc(
			"distribute_api_metrics",
			"Distribute Api Metrics",
			[]string{"app_type", "url", "_type"},
			prometheus.Labels{},
		),
		ShareDataDesc: prometheus.NewDesc(
			"share_data_metrics",
			"Share Data Metrics",
			[]string{"app_type", "_type", "source", "catg"},
			prometheus.Labels{},
		),
		ShareOperationDesc: prometheus.NewDesc(
			"share_operation_metrics",
			"Share Operation Metrics",
			[]string{"app_type", "_type"},
			prometheus.Labels{},
		),
	})

	http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
		promhttp.HandlerFor(prometheus.Gatherers{reg}, promhttp.HandlerOpts{}).ServeHTTP(w, r)
	})
	log.Println("pq Collector Start server at :10000")
	if err := http.ListenAndServe(":10000", nil); err != nil {
		log.Printf("Error occur when start server %v", err)
		os.Exit(1)
	}
}