Commit 31d25dca authored by wuerqiQs's avatar wuerqiQs

cdp -> base all

parent c20c223e
Pipeline #18954 failed with stages
...@@ -16,8 +16,8 @@ import ( ...@@ -16,8 +16,8 @@ import (
"github.com/labstack/echo/v4" "github.com/labstack/echo/v4"
"github.com/spf13/viper" "github.com/spf13/viper"
cmq "github.com/yougg/cmq-go-tdmq" cmq "github.com/yougg/cmq-go-tdmq"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/crypt" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/crypt"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat"
"gopkg.in/gomail.v2" "gopkg.in/gomail.v2"
) )
......
...@@ -5,7 +5,7 @@ import ( ...@@ -5,7 +5,7 @@ import (
"github.com/getsentry/sentry-go" "github.com/getsentry/sentry-go"
"github.com/spf13/viper" "github.com/spf13/viper"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/version" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/version"
) )
var host, _ = os.Hostname() var host, _ = os.Hostname()
......
...@@ -10,7 +10,7 @@ import ( ...@@ -10,7 +10,7 @@ import (
rotatelogs "github.com/lestrrat-go/file-rotatelogs" rotatelogs "github.com/lestrrat-go/file-rotatelogs"
"github.com/rs/zerolog" "github.com/rs/zerolog"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/exception" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/exception"
"go.uber.org/zap" "go.uber.org/zap"
"go.uber.org/zap/zapcore" "go.uber.org/zap/zapcore"
"gopkg.in/natefinch/lumberjack.v2" "gopkg.in/natefinch/lumberjack.v2"
......
...@@ -7,4 +7,4 @@ if [ -x "$(command -v git)" ];then ...@@ -7,4 +7,4 @@ if [ -x "$(command -v git)" ];then
fi fi
basepath=$(cd `dirname $0`; pwd) basepath=$(cd `dirname $0`; pwd)
CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -tags netgo -ldflags "-X 'gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/version.GitHash=$(git show -s --format=%h)' -X 'gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/version.Version=1.0.0'" CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -tags netgo -ldflags "-X 'gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/version.GitHash=$(git show -s --format=%h)' -X 'gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/version.Version=1.0.0'"
...@@ -17,7 +17,7 @@ package cmd ...@@ -17,7 +17,7 @@ package cmd
import ( import (
"github.com/spf13/cobra" "github.com/spf13/cobra"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/exception" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/exception"
) )
// mailCmd represents the mail command // mailCmd represents the mail command
......
...@@ -20,8 +20,8 @@ import ( ...@@ -20,8 +20,8 @@ import (
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
"github.com/spf13/cobra" "github.com/spf13/cobra"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/exception" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/exception"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat"
// mysql // mysql
_ "github.com/go-sql-driver/mysql" _ "github.com/go-sql-driver/mysql"
......
...@@ -20,8 +20,8 @@ import ( ...@@ -20,8 +20,8 @@ import (
"os" "os"
"github.com/spf13/cobra" "github.com/spf13/cobra"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/exception" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/exception"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/version" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/version"
"github.com/spf13/viper" "github.com/spf13/viper"
) )
......
package cmd package cmd
import ( import (
"github.com/spf13/cobra" "github.com/spf13/cobra"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/exception" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/exception"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/supervisor" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/supervisor"
) )
/* /*
由于Supervirsor 通过stdin/stdout 和EventListener 进行协议通信,所以EventListener 不能通过stdout输入其他非协议内容,不然EventListener的状态就会不正常 由于Supervirsor 通过stdin/stdout 和EventListener 进行协议通信,所以EventListener 不能通过stdout输入其他非协议内容,不然EventListener的状态就会不正常
这个暂时还没找到解决方案,暂时把initConfig里的输出注释掉 这个暂时还没找到解决方案,暂时把initConfig里的输出注释掉
*/ */
// mailCmd represents the mail command // mailCmd represents the mail command
var spListenerCmd = &cobra.Command{ var spListenerCmd = &cobra.Command{
Use: "splisten", Use: "splisten",
Short: "supervisor 进程状态监听报警", Short: "supervisor 进程状态监听报警",
Long: `supervisor 进程状态监听报警`, Long: `supervisor 进程状态监听报警`,
Run: func(cmd *cobra.Command, args []string) { Run: func(cmd *cobra.Command, args []string) {
exception.SetAlertViper() exception.SetAlertViper()
supervisor.NewListener().Start() supervisor.NewListener().Start()
}, },
} }
func init() { func init() {
rootCmd.AddCommand(spListenerCmd) rootCmd.AddCommand(spListenerCmd)
} }
...@@ -17,8 +17,8 @@ package cmd ...@@ -17,8 +17,8 @@ package cmd
import ( import (
"github.com/spf13/cobra" "github.com/spf13/cobra"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/exception" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/exception"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/sdktool/tkserver" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/sdktool/tkserver"
) )
// tkserverCmd represents the tkserver command // tkserverCmd represents the tkserver command
......
...@@ -15,7 +15,7 @@ limitations under the License. ...@@ -15,7 +15,7 @@ limitations under the License.
*/ */
package main package main
import "gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/sdktool/cmd" import "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/sdktool/cmd"
func main() { func main() {
cmd.Execute() cmd.Execute()
......
...@@ -8,10 +8,10 @@ import ( ...@@ -8,10 +8,10 @@ import (
"github.com/douyu/jupiter/pkg/server/xgrpc" "github.com/douyu/jupiter/pkg/server/xgrpc"
"github.com/getsentry/sentry-go" "github.com/getsentry/sentry-go"
"github.com/spf13/viper" "github.com/spf13/viper"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/exception" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/exception"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat/cache" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat/cache"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat/client" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat/client"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat/crypt" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat/crypt"
) )
type tkServerEngine struct { type tkServerEngine struct {
......
package supervisor package supervisor
import ( import (
"bufio" "bufio"
"fmt" "fmt"
"io" "io"
"os" "os"
"os/exec" "os/exec"
"regexp" "regexp"
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
"syscall" "syscall"
"time" "time"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/schedule" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/schedule"
"go.uber.org/multierr" "go.uber.org/multierr"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/exception" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/exception"
) )
const ( const (
READY = "READY\n" READY = "READY\n"
RESULTOK = "RESULT 2\nOK" RESULTOK = "RESULT 2\nOK"
) )
type deamon struct { type deamon struct {
pid string pid string
pname string pname string
nextAlert int64 nextAlert int64
alertTime int alertTime int
} }
func (d *deamon) resetState() { func (d *deamon) resetState() {
d.nextAlert = 0 d.nextAlert = 0
d.alertTime = 0 d.alertTime = 0
} }
func (d *deamon) alert() { func (d *deamon) alert() {
now := time.Now().Unix() now := time.Now().Unix()
if d.nextAlert < now { if d.nextAlert < now {
log(fmt.Sprintf("进程:pid:%s pname:%s 未存活", d.pid, d.pname)) log(fmt.Sprintf("进程:pid:%s pname:%s 未存活", d.pid, d.pname))
exception.SendAlert("supervisor_listener", fmt.Errorf("进程:pid:%s pname:%s 未存活", d.pid, d.pname)) exception.SendAlert("supervisor_listener", fmt.Errorf("进程:pid:%s pname:%s 未存活", d.pid, d.pname))
d.alertTime += 1 d.alertTime += 1
if d.alertTime < 20 { if d.alertTime < 20 {
d.nextAlert = now + 60 d.nextAlert = now + 60
} else if d.alertTime < 50 { } else if d.alertTime < 50 {
d.nextAlert = now + 300 d.nextAlert = now + 300
} else if d.alertTime < 100 { } else if d.alertTime < 100 {
d.nextAlert = now + 1800 d.nextAlert = now + 1800
} else { } else {
d.nextAlert = now + 7200 d.nextAlert = now + 7200
} }
} }
log(fmt.Sprintf("%s %s 下次报警时间:%d", d.pid, d.pname, d.nextAlert)) log(fmt.Sprintf("%s %s 下次报警时间:%d", d.pid, d.pname, d.nextAlert))
} }
func NewListener() *Listener { func NewListener() *Listener {
return &Listener{ return &Listener{
deamons: make(map[string]*deamon, 0), deamons: make(map[string]*deamon, 0),
} }
} }
//Listener supervisor 子进程状态监听 //Listener supervisor 子进程状态监听
type Listener struct { type Listener struct {
deamons map[string]*deamon deamons map[string]*deamon
lock sync.Mutex lock sync.Mutex
} }
func log(s string) { func log(s string) {
fmt.Fprintf(os.Stderr, fmt.Sprintf("info:%s\n", s)) fmt.Fprintf(os.Stderr, fmt.Sprintf("info:%s\n", s))
os.Stderr.Sync() os.Stderr.Sync()
} }
func (sl *Listener) state(s string) { func (sl *Listener) state(s string) {
fmt.Fprintf(os.Stdout, s) fmt.Fprintf(os.Stdout, s)
os.Stdout.Sync() os.Stdout.Sync()
} }
func (sl *Listener) registerDeamon(pname string, pid string) { func (sl *Listener) registerDeamon(pname string, pid string) {
sl.lock.Lock() sl.lock.Lock()
defer sl.lock.Unlock() defer sl.lock.Unlock()
sl.deamons[pname] = &deamon{pid: pid, pname: pname, nextAlert: 0, alertTime: 0} sl.deamons[pname] = &deamon{pid: pid, pname: pname, nextAlert: 0, alertTime: 0}
} }
func (sl *Listener) unRegisterDeamon(pname string, pid string) { func (sl *Listener) unRegisterDeamon(pname string, pid string) {
if _, ok := sl.deamons[pname]; ok { if _, ok := sl.deamons[pname]; ok {
sl.lock.Lock() sl.lock.Lock()
defer sl.lock.Unlock() defer sl.lock.Unlock()
delete(sl.deamons, pname) delete(sl.deamons, pname)
} }
} }
func (sl *Listener) init() { func (sl *Listener) init() {
cmd := exec.Command("supervisorctl", "status", "") cmd := exec.Command("supervisorctl", "status", "")
out, err := cmd.Output() out, err := cmd.Output()
if err != nil { if err != nil {
log(err.Error()) log(err.Error())
} }
outlines := strings.Split(string(out), "\n") outlines := strings.Split(string(out), "\n")
sl.lock.Lock() sl.lock.Lock()
defer sl.lock.Unlock() defer sl.lock.Unlock()
for _, outline := range outlines { for _, outline := range outlines {
spitem := strings.Split(outline, ",") spitem := strings.Split(outline, ",")
pinfo := strings.Split(deleteExtraSpace(spitem[0]), " ") pinfo := strings.Split(deleteExtraSpace(spitem[0]), " ")
if len(pinfo) == 4 { if len(pinfo) == 4 {
if pinfo[1] == "RUNNING" { if pinfo[1] == "RUNNING" {
sl.deamons[pinfo[0]] = &deamon{pid: pinfo[3], pname: pinfo[0], nextAlert: 0, alertTime: 0} sl.deamons[pinfo[0]] = &deamon{pid: pinfo[3], pname: pinfo[0], nextAlert: 0, alertTime: 0}
} }
} }
} }
} }
func (sl *Listener) checkDeamonState() { func (sl *Listener) checkDeamonState() {
tt := schedule.NewTimerTask(schedule.WithErrHandler(func(name string, err error) { tt := schedule.NewTimerTask(schedule.WithErrHandler(func(name string, err error) {
exception.SendAlert("supervisor_listener", fmt.Errorf("%s:%s", name, err.Error())) exception.SendAlert("supervisor_listener", fmt.Errorf("%s:%s", name, err.Error()))
}), schedule.WithRunHandler(func(name string, nextAt time.Time) { }), schedule.WithRunHandler(func(name string, nextAt time.Time) {
// log(name) // log(name)
})) }))
tt.Do(time.Now(), time.Second*10, func(args ...interface{}) error { tt.Do(time.Now(), time.Second*10, func(args ...interface{}) error {
sl.lock.Lock() sl.lock.Lock()
defer sl.lock.Unlock() defer sl.lock.Unlock()
var err error var err error
for _, deamon := range sl.deamons { for _, deamon := range sl.deamons {
pid, e := strconv.Atoi(deamon.pid) pid, e := strconv.Atoi(deamon.pid)
if e != nil { if e != nil {
err = multierr.Append(err, e) err = multierr.Append(err, e)
continue continue
} }
p, err := os.FindProcess(pid) p, err := os.FindProcess(pid)
if err != nil { if err != nil {
deamon.alert() deamon.alert()
} else { } else {
if p == nil { if p == nil {
deamon.alert() deamon.alert()
} else { } else {
if err := p.Signal(syscall.Signal(0)); err != nil { if err := p.Signal(syscall.Signal(0)); err != nil {
deamon.alert() deamon.alert()
} else { } else {
deamon.resetState() deamon.resetState()
} }
} }
} }
} }
return err return err
}, "进程状态监测") }, "进程状态监测")
tt.Run() tt.Run()
} }
func (sl *Listener) recover(process, from, to string) { func (sl *Listener) recover(process, from, to string) {
log(fmt.Sprintf("进程 %s 被拉起", process)) log(fmt.Sprintf("进程 %s 被拉起", process))
exception.SendAlert("supervisor_listener", fmt.Errorf("进程 %s 拉起成功", process)) exception.SendAlert("supervisor_listener", fmt.Errorf("进程 %s 拉起成功", process))
} }
func (sl *Listener) panic(process, from, to string) { func (sl *Listener) panic(process, from, to string) {
log(fmt.Sprintf("进程 %s 异常退出,请注意排查", process)) log(fmt.Sprintf("进程 %s 异常退出,请注意排查", process))
exception.SendAlert("supervisor_listener", fmt.Errorf("进程 %s 异常退出,请注意排查", process)) exception.SendAlert("supervisor_listener", fmt.Errorf("进程 %s 异常退出,请注意排查", process))
} }
func stringSliceContain(slice []string, value string) bool { func stringSliceContain(slice []string, value string) bool {
for _, item := range slice { for _, item := range slice {
if item == value { if item == value {
return true return true
} }
} }
return false return false
} }
func deleteExtraSpace(s string) string { func deleteExtraSpace(s string) string {
s1 := strings.Replace(s, " ", " ", -1) s1 := strings.Replace(s, " ", " ", -1)
regstr := "\\s{2,}" regstr := "\\s{2,}"
reg, _ := regexp.Compile(regstr) reg, _ := regexp.Compile(regstr)
s2 := make([]byte, len(s1)) s2 := make([]byte, len(s1))
copy(s2, s1) copy(s2, s1)
spc_index := reg.FindStringIndex(string(s2)) spc_index := reg.FindStringIndex(string(s2))
for len(spc_index) > 0 { for len(spc_index) > 0 {
s2 = append(s2[:spc_index[0]+1], s2[spc_index[1]:]...) s2 = append(s2[:spc_index[0]+1], s2[spc_index[1]:]...)
spc_index = reg.FindStringIndex(string(s2)) spc_index = reg.FindStringIndex(string(s2))
} }
return string(s2) return string(s2)
} }
//Start 开始监听子进程状态 //Start 开始监听子进程状态
func (sl *Listener) Start() { func (sl *Listener) Start() {
sl.init() sl.init()
sl.checkDeamonState() sl.checkDeamonState()
lastSerial := "0" lastSerial := "0"
lastPoolSerial := "0" lastPoolSerial := "0"
psExit := make(map[string]string, 0) psExit := make(map[string]string, 0)
var body [4096]byte var body [4096]byte
reader := bufio.NewReader(os.Stdin) reader := bufio.NewReader(os.Stdin)
for { for {
sl.state(READY) sl.state(READY)
line, _, err := reader.ReadLine() line, _, err := reader.ReadLine()
if err != nil { if err != nil {
exception.SendAlert("supervisor_listener", err) exception.SendAlert("supervisor_listener", err)
continue continue
} }
log(string(line)) log(string(line))
header := make(map[string]string, 0) header := make(map[string]string, 0)
items := strings.Split(string(line), " ") items := strings.Split(string(line), " ")
for _, item := range items { for _, item := range items {
kv := strings.Split(item, ":") kv := strings.Split(item, ":")
header[kv[0]] = kv[1] header[kv[0]] = kv[1]
} }
serial := header["serial"] serial := header["serial"]
poolSerial := header["poolserial"] poolSerial := header["poolserial"]
bodylen, err := strconv.ParseFloat(header["len"], 32) bodylen, err := strconv.ParseFloat(header["len"], 32)
if err != nil { if err != nil {
exception.SendAlert("supervisor_listener", err) exception.SendAlert("supervisor_listener", err)
continue continue
} }
_, err = io.ReadFull(reader, body[:int(bodylen)]) _, err = io.ReadFull(reader, body[:int(bodylen)])
if err != nil { if err != nil {
exception.SendAlert("supervisor_listener", err) exception.SendAlert("supervisor_listener", err)
continue continue
} }
data := body[:int(bodylen)] data := body[:int(bodylen)]
log("body+" + string(data)) log("body+" + string(data))
if serial == lastSerial && poolSerial == lastPoolSerial { if serial == lastSerial && poolSerial == lastPoolSerial {
sl.state(RESULTOK) sl.state(RESULTOK)
} }
lastSerial = serial lastSerial = serial
lastPoolSerial = poolSerial lastPoolSerial = poolSerial
bodys := make(map[string]string, 0) bodys := make(map[string]string, 0)
items = strings.Split(string(data), " ") items = strings.Split(string(data), " ")
for _, item := range items { for _, item := range items {
kv := strings.Split(item, ":") kv := strings.Split(item, ":")
if len(kv) > 1 { if len(kv) > 1 {
bodys[kv[0]] = kv[1] bodys[kv[0]] = kv[1]
} }
} }
process := bodys["processname"] process := bodys["processname"]
fromState := bodys["from_state"] fromState := bodys["from_state"]
pid := bodys["pid"] pid := bodys["pid"]
toState := header["eventname"] toState := header["eventname"]
log(fmt.Sprintf("\nprocessname:[%s] from_state:[%s] to_state:[%s] last_serial:[%s] last_pool_serial:[%s]\n", process, fromState, toState, lastSerial, lastPoolSerial)) log(fmt.Sprintf("\nprocessname:[%s] from_state:[%s] to_state:[%s] last_serial:[%s] last_pool_serial:[%s]\n", process, fromState, toState, lastSerial, lastPoolSerial))
if stringSliceContain([]string{"PROCESS_STATE_RUNNING"}, toState) { if stringSliceContain([]string{"PROCESS_STATE_RUNNING"}, toState) {
sl.registerDeamon(process, pid) sl.registerDeamon(process, pid)
if _, ok := psExit[process]; ok { if _, ok := psExit[process]; ok {
delete(psExit, process) delete(psExit, process)
sl.recover(process, fromState, toState) sl.recover(process, fromState, toState)
} }
} }
if stringSliceContain([]string{"PROCESS_STATE_STOPPED"}, toState) { if stringSliceContain([]string{"PROCESS_STATE_STOPPED"}, toState) {
sl.unRegisterDeamon(process, pid) sl.unRegisterDeamon(process, pid)
} }
if stringSliceContain([]string{"PROCESS_STATE_EXITED", "PROCESS_STATE_FATAL", "PROCESS_STATE_BACKOFF"}, toState) { if stringSliceContain([]string{"PROCESS_STATE_EXITED", "PROCESS_STATE_FATAL", "PROCESS_STATE_BACKOFF"}, toState) {
psExit[process] = "" psExit[process] = ""
sl.panic(process, fromState, toState) sl.panic(process, fromState, toState)
} }
sl.state(RESULTOK) sl.state(RESULTOK)
} }
} }
package client package client
import ( import (
"bytes" "bytes"
"encoding/json" "encoding/json"
"fmt" "fmt"
"io" "io"
"io/ioutil" "io/ioutil"
"net/http" "net/http"
"net/url" "net/url"
"time" "time"
"github.com/glutwins/webclient" "github.com/glutwins/webclient"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat/cache" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat/cache"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat/crypt" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat/crypt"
) )
type WachatReq map[string]interface{} type WachatReq map[string]interface{}
type Client struct { type Client struct {
*crypt.WechatConfig *crypt.WechatConfig
data cache.Cache data cache.Cache
tkexpire bool tkexpire bool
} }
func NewClient(cfg *crypt.WechatConfig, d cache.Cache) *Client { func NewClient(cfg *crypt.WechatConfig, d cache.Cache) *Client {
c := &Client{WechatConfig: cfg, data: d} c := &Client{WechatConfig: cfg, data: d}
return c return c
} }
type JsConfig struct { type JsConfig struct {
AppID string AppID string
TimeStamp int64 TimeStamp int64
NonceStr string NonceStr string
Signature string Signature string
} }
type cacheClient struct { type cacheClient struct {
Token accessToken Token accessToken
Ticket resTicket Ticket resTicket
} }
type tokenReq struct { type tokenReq struct {
Sign string `json:"Auth-Sign"` Sign string `json:"Auth-Sign"`
OverdueToken string `json:"overdue_token"` OverdueToken string `json:"overdue_token"`
Eid int `json:"enterprise_id"` Eid int `json:"enterprise_id"`
} }
type tokenRsp struct { type tokenRsp struct {
Token string `json:"access_token"` Token string `json:"access_token"`
} }
//GetAccessToken 获取access_token //GetAccessToken 获取access_token
func (c *Client) GetAccessToken() (string, error) { func (c *Client) GetAccessToken() (string, error) {
var token accessToken var token accessToken
if !c.tkexpire { if !c.tkexpire {
if err := c.data.Get("token", &token); err == nil { if err := c.data.Get("token", &token); err == nil {
if token.valid() { if token.valid() {
return token.AccessToken, nil return token.AccessToken, nil
} }
} }
} }
c.data.Lock("token") c.data.Lock("token")
defer c.data.Unlock("token") defer c.data.Unlock("token")
url := fmt.Sprintf(accessTokenURL, c.AppID, c.AppSecret) url := fmt.Sprintf(accessTokenURL, c.AppID, c.AppSecret)
resp, err := http.Get(url) resp, err := http.Get(url)
if err != nil { if err != nil {
return "", err return "", err
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("GetAccessToken httpcode=%d", resp.StatusCode) return "", fmt.Errorf("GetAccessToken httpcode=%d", resp.StatusCode)
} }
if b, err := ioutil.ReadAll(resp.Body); err != nil { if b, err := ioutil.ReadAll(resp.Body); err != nil {
return "", err return "", err
} else if err = json.Unmarshal(b, &token); err != nil { } else if err = json.Unmarshal(b, &token); err != nil {
return "", err return "", err
} }
token.LastUpTs = time.Now().Unix() token.LastUpTs = time.Now().Unix()
c.data.Set("token", token) c.data.Set("token", token)
c.tkexpire = false c.tkexpire = false
return token.AccessToken, token.Error() return token.AccessToken, token.Error()
} }
// ExpireToken 使token失效 // ExpireToken 使token失效
func (c *Client) ExpireToken() { func (c *Client) ExpireToken() {
c.tkexpire = true c.tkexpire = true
} }
func (c *Client) formatUrlWithAccessToken(base string, args ...interface{}) (string, error) { func (c *Client) formatUrlWithAccessToken(base string, args ...interface{}) (string, error) {
var token string var token string
var err error var err error
token, err = c.GetAccessToken() token, err = c.GetAccessToken()
if err != nil { if err != nil {
return "", err return "", err
} }
return fmt.Sprintf(base, append([]interface{}{token}, args...)...), nil return fmt.Sprintf(base, append([]interface{}{token}, args...)...), nil
} }
func (c *Client) postJsonUrlFormat(req interface{}, res interface{}, url string, args ...interface{}) error { func (c *Client) postJsonUrlFormat(req interface{}, res interface{}, url string, args ...interface{}) error {
if uri, err := c.formatUrlWithAccessToken(url, args...); err != nil { if uri, err := c.formatUrlWithAccessToken(url, args...); err != nil {
return err return err
} else if response, err := c.jsonPost(uri, req); err != nil { } else if response, err := c.jsonPost(uri, req); err != nil {
return err return err
} else if err := json.Unmarshal(response, res); err != nil { } else if err := json.Unmarshal(response, res); err != nil {
return err return err
} }
return nil return nil
} }
func (c *Client) jsonPost(uri string, obj interface{}) ([]byte, error) { func (c *Client) jsonPost(uri string, obj interface{}) ([]byte, error) {
var jsonData []byte var jsonData []byte
var err error var err error
if str, ok := obj.(string); ok { if str, ok := obj.(string); ok {
jsonData = []byte(str) jsonData = []byte(str)
} else { } else {
jsonData, err = json.Marshal(obj) jsonData, err = json.Marshal(obj)
if err != nil { if err != nil {
return nil, err return nil, err
} }
} }
return c.doPost(uri, "application/json;charset=utf-8", bytes.NewBuffer(jsonData)) return c.doPost(uri, "application/json;charset=utf-8", bytes.NewBuffer(jsonData))
} }
func (c *Client) formPost(uri string, data map[string]string) ([]byte, error) { func (c *Client) formPost(uri string, data map[string]string) ([]byte, error) {
body := bytes.NewBuffer(nil) body := bytes.NewBuffer(nil)
for k, v := range data { for k, v := range data {
body.WriteString(k) body.WriteString(k)
body.WriteString("=") body.WriteString("=")
body.WriteString(url.QueryEscape(v)) body.WriteString(url.QueryEscape(v))
body.WriteString("&") body.WriteString("&")
} }
body.Truncate(body.Len() - 1) body.Truncate(body.Len() - 1)
return c.doPost(uri, "application/x-www-form-urlencoded;charset=utf-8", body) return c.doPost(uri, "application/x-www-form-urlencoded;charset=utf-8", body)
} }
func (c *Client) doPost(uri, contentType string, r io.Reader) ([]byte, error) { func (c *Client) doPost(uri, contentType string, r io.Reader) ([]byte, error) {
response, err := http.Post(uri, contentType, r) response, err := http.Post(uri, contentType, r)
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer response.Body.Close() defer response.Body.Close()
if response.StatusCode != http.StatusOK { if response.StatusCode != http.StatusOK {
return nil, fmt.Errorf("http get error : uri=%v , statusCode=%v", uri, response.StatusCode) return nil, fmt.Errorf("http get error : uri=%v , statusCode=%v", uri, response.StatusCode)
} }
return ioutil.ReadAll(response.Body) return ioutil.ReadAll(response.Body)
} }
func (c *Client) getJsonUrlFormat(res interface{}, url string, args ...interface{}) error { func (c *Client) getJsonUrlFormat(res interface{}, url string, args ...interface{}) error {
uri, err := c.formatUrlWithAccessToken(url, args...) uri, err := c.formatUrlWithAccessToken(url, args...)
if err != nil { if err != nil {
return err return err
} }
if b, err := webclient.DoGet(uri); err != nil { if b, err := webclient.DoGet(uri); err != nil {
return err return err
} else { } else {
return json.Unmarshal(b, res) return json.Unmarshal(b, res)
} }
} }
//uri 为当前网页地址 //uri 为当前网页地址
func (js *Client) GetJsConfig(uri string) (config *JsConfig, err error) { func (js *Client) GetJsConfig(uri string) (config *JsConfig, err error) {
config = new(JsConfig) config = new(JsConfig)
var ticketStr string var ticketStr string
ticketStr, err = js.getTicket() ticketStr, err = js.getTicket()
if err != nil { if err != nil {
return return
} }
nonceStr := crypt.RandomStr(16) nonceStr := crypt.RandomStr(16)
timestamp := time.Now().Unix() timestamp := time.Now().Unix()
str := fmt.Sprintf("jsapi_ticket=%s&noncestr=%s&timestamp=%d&url=%s", ticketStr, nonceStr, timestamp, uri) str := fmt.Sprintf("jsapi_ticket=%s&noncestr=%s&timestamp=%d&url=%s", ticketStr, nonceStr, timestamp, uri)
sigStr := crypt.Signature(str) sigStr := crypt.Signature(str)
config.AppID = js.AppID config.AppID = js.AppID
config.NonceStr = nonceStr config.NonceStr = nonceStr
config.TimeStamp = timestamp config.TimeStamp = timestamp
config.Signature = sigStr config.Signature = sigStr
return return
} }
//getTicket 获取jsapi_tocket全局缓存 //getTicket 获取jsapi_tocket全局缓存
func (c *Client) getTicket() (string, error) { func (c *Client) getTicket() (string, error) {
var ticket resTicket var ticket resTicket
if err := c.data.Get("ticket", &ticket); err == nil { if err := c.data.Get("ticket", &ticket); err == nil {
if ticket.valid() { if ticket.valid() {
return ticket.Ticket, nil return ticket.Ticket, nil
} }
} }
c.data.Lock("ticket") c.data.Lock("ticket")
defer c.data.Unlock("ticket") defer c.data.Unlock("ticket")
if err := c.getJsonUrlFormat(&ticket, getTicketURL); err != nil { if err := c.getJsonUrlFormat(&ticket, getTicketURL); err != nil {
return "", err return "", err
} }
ticket.LastUpTs = time.Now().Unix() ticket.LastUpTs = time.Now().Unix()
c.data.Set("ticket", ticket) c.data.Set("ticket", ticket)
return ticket.Ticket, ticket.Error() return ticket.Ticket, ticket.Error()
} }
func (c *Client) NewQrCode(sceneId interface{}, expire int) (*QrCode, error) { func (c *Client) NewQrCode(sceneId interface{}, expire int) (*QrCode, error) {
var msg string var msg string
if _, ok := sceneId.(string); ok { if _, ok := sceneId.(string); ok {
if expire == 0 { if expire == 0 {
msg = fmt.Sprintf(kQrStrLimitFormat, sceneId) msg = fmt.Sprintf(kQrStrLimitFormat, sceneId)
} else { } else {
return nil, fmt.Errorf("not support expire str qrcode") return nil, fmt.Errorf("not support expire str qrcode")
} }
} else { } else {
if expire == 0 { if expire == 0 {
msg = fmt.Sprintf(kQrIntLimitFormat, sceneId) msg = fmt.Sprintf(kQrIntLimitFormat, sceneId)
} else { } else {
msg = fmt.Sprintf(kQrIntFormat, expire, sceneId) msg = fmt.Sprintf(kQrIntFormat, expire, sceneId)
} }
} }
var qrcode = &QrCode{} var qrcode = &QrCode{}
if err := c.postJsonUrlFormat(msg, qrcode, qrcodeURL); err != nil { if err := c.postJsonUrlFormat(msg, qrcode, qrcodeURL); err != nil {
return nil, err return nil, err
} }
return qrcode, qrcode.Error() return qrcode, qrcode.Error()
} }
type CustomerText struct { type CustomerText struct {
Content string `json:"content"` Content string `json:"content"`
} }
type CustomerImage struct { type CustomerImage struct {
MediaID string `json:"media_id"` MediaID string `json:"media_id"`
} }
type CustomerMessage struct { type CustomerMessage struct {
ToUser string `json:"touser"` ToUser string `json:"touser"`
MsgType string `json:"msgtype"` MsgType string `json:"msgtype"`
Text *CustomerText `json:"text,omitempty"` Text *CustomerText `json:"text,omitempty"`
Image *CustomerImage `json:"image,omitempty"` Image *CustomerImage `json:"image,omitempty"`
} }
func (c *Client) SendMessage(msg *CustomerMessage) error { func (c *Client) SendMessage(msg *CustomerMessage) error {
resp := &CommonResp{} resp := &CommonResp{}
if err := c.postJsonUrlFormat(msg, resp, customerMsgURL); err != nil { if err := c.postJsonUrlFormat(msg, resp, customerMsgURL); err != nil {
return err return err
} }
if resp.ErrCode == 40001 { if resp.ErrCode == 40001 {
if err := c.postJsonUrlFormat(msg, resp, customerMsgURL); err != nil { if err := c.postJsonUrlFormat(msg, resp, customerMsgURL); err != nil {
return err return err
} }
return resp.Error() return resp.Error()
} }
return resp.Error() return resp.Error()
} }
package context package context
import ( import (
"encoding/xml" "encoding/xml"
"errors" "errors"
"io/ioutil" "io/ioutil"
"net/http" "net/http"
"net/url" "net/url"
"strconv" "strconv"
"time" "time"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat/crypt" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat/crypt"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat/message" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat/message"
) )
var ( var (
// ErrSign wechat sign error // ErrSign wechat sign error
ErrSign = errors.New("invalid sign") ErrSign = errors.New("invalid sign")
) )
type Context interface { type Context interface {
Query(key string) (string, bool) Query(key string) (string, bool)
ParseRawMessage() (*message.MixMessage, error) ParseRawMessage() (*message.MixMessage, error)
Render([]byte) Render([]byte)
RenderString(string) RenderString(string)
RenderXML(interface{}) RenderXML(interface{})
RenderMessage(message.Reply) RenderMessage(message.Reply)
} }
func NewContext(req *http.Request, w http.ResponseWriter, cfg *crypt.WechatConfig) Context { func NewContext(req *http.Request, w http.ResponseWriter, cfg *crypt.WechatConfig) Context {
ctx := &ctxImpl{Writer: w, Request: req, cfg: cfg} ctx := &ctxImpl{Writer: w, Request: req, cfg: cfg}
ctx.parse() ctx.parse()
return ctx return ctx
} }
// Context struct // Context struct
type ctxImpl struct { type ctxImpl struct {
Writer http.ResponseWriter Writer http.ResponseWriter
Request *http.Request Request *http.Request
isSafeMode bool isSafeMode bool
params url.Values params url.Values
random []byte random []byte
cfg *crypt.WechatConfig cfg *crypt.WechatConfig
} }
func (ctx *ctxImpl) parse() { func (ctx *ctxImpl) parse() {
ctx.params = ctx.Request.URL.Query() ctx.params = ctx.Request.URL.Query()
if enctype, ok := ctx.Query("encrypt_type"); ok { if enctype, ok := ctx.Query("encrypt_type"); ok {
ctx.isSafeMode = enctype == "aes" ctx.isSafeMode = enctype == "aes"
} }
} }
func (ctx *ctxImpl) ParseRawMessage() (*message.MixMessage, error) { func (ctx *ctxImpl) ParseRawMessage() (*message.MixMessage, error) {
timestamp, _ := ctx.Query("timestamp") timestamp, _ := ctx.Query("timestamp")
nonce, _ := ctx.Query("nonce") nonce, _ := ctx.Query("nonce")
signature, _ := ctx.Query("signature") signature, _ := ctx.Query("signature")
if signature != crypt.Signature(ctx.cfg.Token, timestamp, nonce) { if signature != crypt.Signature(ctx.cfg.Token, timestamp, nonce) {
return nil, ErrSign return nil, ErrSign
} }
var rawXMLMsgBytes []byte var rawXMLMsgBytes []byte
var err error var err error
if ctx.isSafeMode { if ctx.isSafeMode {
var encryptedXMLMsg message.EncryptedXMLMsg var encryptedXMLMsg message.EncryptedXMLMsg
if err = xml.NewDecoder(ctx.Request.Body).Decode(&encryptedXMLMsg); err != nil { if err = xml.NewDecoder(ctx.Request.Body).Decode(&encryptedXMLMsg); err != nil {
return nil, err return nil, err
} }
msgSignature, _ := ctx.Query("msg_signature") msgSignature, _ := ctx.Query("msg_signature")
msgSignatureGen := crypt.Signature(ctx.cfg.Token, timestamp, nonce, encryptedXMLMsg.EncryptedMsg) msgSignatureGen := crypt.Signature(ctx.cfg.Token, timestamp, nonce, encryptedXMLMsg.EncryptedMsg)
if msgSignature != msgSignatureGen { if msgSignature != msgSignatureGen {
return nil, ErrSign return nil, ErrSign
} }
//解密 //解密
ctx.random, rawXMLMsgBytes, err = crypt.DecryptMsg(ctx.cfg.AppID, encryptedXMLMsg.EncryptedMsg, ctx.cfg.AESKey) ctx.random, rawXMLMsgBytes, err = crypt.DecryptMsg(ctx.cfg.AppID, encryptedXMLMsg.EncryptedMsg, ctx.cfg.AESKey)
if err != nil { if err != nil {
return nil, err return nil, err
} }
} else { } else {
rawXMLMsgBytes, err = ioutil.ReadAll(ctx.Request.Body) rawXMLMsgBytes, err = ioutil.ReadAll(ctx.Request.Body)
if err != nil { if err != nil {
return nil, err return nil, err
} }
} }
msg := &message.MixMessage{} msg := &message.MixMessage{}
err = xml.Unmarshal(rawXMLMsgBytes, msg) err = xml.Unmarshal(rawXMLMsgBytes, msg)
return msg, err return msg, err
} }
// GetQuery is like Query(), it returns the keyed url query value // GetQuery is like Query(), it returns the keyed url query value
func (ctx *ctxImpl) Query(key string) (string, bool) { func (ctx *ctxImpl) Query(key string) (string, bool) {
if values, ok := ctx.params[key]; ok && len(values) > 0 { if values, ok := ctx.params[key]; ok && len(values) > 0 {
return values[0], true return values[0], true
} }
return "", false return "", false
} }
//Render render from bytes //Render render from bytes
func (ctx *ctxImpl) Render(bytes []byte) { func (ctx *ctxImpl) Render(bytes []byte) {
ctx.Writer.WriteHeader(200) ctx.Writer.WriteHeader(200)
_, err := ctx.Writer.Write(bytes) _, err := ctx.Writer.Write(bytes)
if err != nil { if err != nil {
panic(err) panic(err)
} }
} }
//String render from string //String render from string
func (ctx *ctxImpl) RenderString(str string) { func (ctx *ctxImpl) RenderString(str string) {
ctx.Writer.Header().Set("Content-Type", "text/plain; charset=utf-8") ctx.Writer.Header().Set("Content-Type", "text/plain; charset=utf-8")
ctx.Render([]byte(str)) ctx.Render([]byte(str))
} }
//XML render to xml //XML render to xml
func (ctx *ctxImpl) RenderXML(obj interface{}) { func (ctx *ctxImpl) RenderXML(obj interface{}) {
ctx.Writer.Header().Set("Content-Type", "application/xml; charset=utf-8") ctx.Writer.Header().Set("Content-Type", "application/xml; charset=utf-8")
bytes, err := xml.Marshal(obj) bytes, err := xml.Marshal(obj)
if err != nil { if err != nil {
panic(err) panic(err)
} }
ctx.Render(bytes) ctx.Render(bytes)
} }
func (ctx *ctxImpl) RenderMessage(reply message.Reply) { func (ctx *ctxImpl) RenderMessage(reply message.Reply) {
var replyMsg interface{} = reply var replyMsg interface{} = reply
if ctx.isSafeMode { if ctx.isSafeMode {
rawMsg, err := xml.Marshal(replyMsg) rawMsg, err := xml.Marshal(replyMsg)
if err != nil { if err != nil {
return return
} }
//安全模式下对消息进行加密 //安全模式下对消息进行加密
encryptedMsg, err := crypt.EncryptMsg(ctx.random, rawMsg, ctx.cfg.AppID, ctx.cfg.AESKey) encryptedMsg, err := crypt.EncryptMsg(ctx.random, rawMsg, ctx.cfg.AppID, ctx.cfg.AESKey)
if err != nil { if err != nil {
return return
} }
timestamp := time.Now().Unix() timestamp := time.Now().Unix()
strts := strconv.FormatInt(timestamp, 64) strts := strconv.FormatInt(timestamp, 64)
nonce := crypt.RandomStr(16) nonce := crypt.RandomStr(16)
msgSignature := crypt.Signature(ctx.cfg.Token, strts, nonce, string(encryptedMsg)) msgSignature := crypt.Signature(ctx.cfg.Token, strts, nonce, string(encryptedMsg))
replyMsg = message.ResponseEncryptedXMLMsg{ replyMsg = message.ResponseEncryptedXMLMsg{
EncryptedMsg: string(encryptedMsg), EncryptedMsg: string(encryptedMsg),
MsgSignature: msgSignature, MsgSignature: msgSignature,
Timestamp: timestamp, Timestamp: timestamp,
Nonce: nonce, Nonce: nonce,
} }
} }
ctx.RenderXML(replyMsg) ctx.RenderXML(replyMsg)
return return
} }
...@@ -5,7 +5,7 @@ import ( ...@@ -5,7 +5,7 @@ import (
"encoding/xml" "encoding/xml"
"io/ioutil" "io/ioutil"
"gitlab-ce.k8s.tools.vchangyi.com/cdp/cy-sdk-go/wechat/crypt" "gitlab-ce.k8s.tools.vchangyi.com/base/cy-sdk-go/wechat/crypt"
) )
type MchBaseRes struct { type MchBaseRes struct {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment