0
点赞
收藏
分享

微信扫一扫

获取k8s中pod日志上传oss输出到企业微信

获取k8s中pod日志上传oss输出到企业微信
//目的: 一键获取Pod日志,方便给其他人查看
//方式:
//1.通过client-go获取pod日志
//2.将pod日志上传值oss
//3.将返回的链接输出到企业微信
//注意点:
//1.参考kubectl log源码去写的
//2.上传值Oss信息PutOSSFile函数我涂抹了,需要自己写

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "errors"
    "flag"
    "fmt"
    "github.com/aliyun/aliyun-oss-go-sdk/oss"
    "io"
    "io/ioutil"
    "k8s.io/client-go/util/homedir"
    "log"
    "math"
    "net/http"
    "os"
    "path/filepath"
    "strings"

    apiv1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/rest"
    "k8s.io/client-go/tools/clientcmd"
    "time"
)

//企业微信数据结构定义markdown
type MessageMD struct {
    MsgType  string `json:"msgtype"`
    Markdown struct {
        Content        string   `json:"content"`
        Mentioned_list []string `json:"mentioned_list"`
    } `json:"markdown"`
}

//参数列表,目的是为了根据需求进行排除,例如不监控特定namespace和特定ns空间的信息
var (
    url        = flag.String("url", "", "请输入企业微信webhook地址")
    ns         = flag.String("ns", "", "请输入命名空间名称,n1,n2")
    n          = flag.String("n", "", "请输入Pod名称,可以只写deployment名称,n1,n2")
    follow     = flag.Bool("f", false, "是否需要指定日志为流模式")
    timestamps = flag.Bool("t", false, "是否输出时间戳")
    tail       = flag.Int64("l", -1, "请输入想显示最后多少行数据")
    sincetime  = flag.String("st", "", "请输入日志起始时间,时间格式为2021-12-31T11:10:19")
    since      = flag.String("s", "", "请指定某段时间内的日志, 5s, 2m, or 3h")
    container  = flag.String("c", "", "请指定pod中容器名称")
    kubeconfig = flag.String("k", "", "请输入kubeconfig路径")
)

//设置logOption
var logOptions = &apiv1.PodLogOptions{}

func main() {
    //默认参数解析
    flag.Parse()
    //设置默认logOption
    SetlogOption()
    //获取日志主函数
    GetPodLog()

}

//资源监控主函数
func GetPodLog() {
    var err error
    var config *rest.Config
    var pod apiv1.Pod
    var podIsExist bool

    // 初始化 rest.Config 对象
    if config, err = rest.InClusterConfig(); err != nil {
        if config, err = clientcmd.BuildConfigFromFlags("", *kubeconfig); err != nil {
            CheckError("kubeconfig配置文件出错: ", err)
        }
    }
    // 创建 Clientset 对象
    clientset, err := kubernetes.NewForConfig(config)
    CheckError("获取clientset出错:", err)

    ctx := context.Background()
    list, err := clientset.CoreV1().Pods(*ns).List(ctx, metav1.ListOptions{})
    CheckError("获取Podlist出错:", err)
    for _, v := range list.Items {
        //暂时返回第一个匹配的进行处理,还没想好怎么解决
        if strings.Contains(v.Name, *n) {
            pod = v
            nsn := v.Namespace + "/" + v.Name
            fmt.Println("该pod名称:", nsn)
            podIsExist = true
            break
        }
    }
    if !podIsExist {
        s := "命名空间:" + *ns
        CheckError(s, errors.New("没有这个Pod,或者Pod名输入错误!"))
    }

    //设置默认获取第一个容器地址
    if len(*container) == 0 {
        // [CONTAINER] (container as arg not flag) is supported as legacy behavior. See PR #10519 for more details.
        if len(pod.Spec.Containers) != 1 {
            podContainersNames := []string{}
            for _, container := range pod.Spec.Containers {
                podContainersNames = append(podContainersNames, container.Name)
            }
        } else {
            container = &pod.Spec.Containers[0].Name
        }
    }
    //这里container获取contanier的值
    logOptions.Container = *container

    //真正获取pod的日志了
    stream, err := clientset.CoreV1().Pods(*ns).GetLogs(pod.Name, logOptions).Stream(context.TODO())
    CheckError("打开pod日志出错", err)

    defer stream.Close()
    buf := new(bytes.Buffer)
    _, err = io.Copy(buf, stream)
    CheckError("拷贝pod日志信息出错", err)
    str := buf.String()
    //fmt.Println(str)

    //上传至oss返回地址
    logOss, err := PutOSSFile(pod.Namespace, pod.Name, str)
    CheckError("上传Oss地址出错", err)

    timenow := time.Now().Format("2006-01-02 15:04:05")
    urladdr := "[链接地址](" + logOss + ")"
    context := "Pod日志获取信息:" + "\nPod名称:  " + "**" + pod.Name + "**" + "\n命名空间:  " + pod.Namespace + "\n现在时间:  " + timenow + "\n日志地址: " + urladdr

    //设置默认webhoo地址
    if len(*url) == 0 {
        CheckError("请输入webhook地址",errors.New(""))
        //url1 := "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxx"
        //url = &url1
    }
    SendMessage(*url, context)
}

//上传pod文件
func PutOSSFile(namespace, podname, body string) (string, error) {
    timenow := time.Now().Format("2006-01-02_15-04-05")
    endpoint := "xxx"
    accessKeyId := "yyy"
    accessKeySecret := "zzz"
    bucketName := "ddd"
    client1 := &http.Client{
        Transport: &http.Transport{
            DisableKeepAlives: true,
        },
    }
    client, err := oss.New(endpoint, accessKeyId, accessKeySecret, oss.HTTPClient(client1))
    bucket, err := client.Bucket(bucketName)
    if err != nil {
        return "", err
    }
    objectName := "eee/" + namespace + "-" + podname + "_" + timenow + ".log"
    err = bucket.PutObject(objectName, bytes.NewReader([]byte(body)))
    if err != nil {
        return "", err
    }
    downloadurl := `https://ffff/` + objectName
    return downloadurl, nil
}

//错误检查,以及退出
func CheckError(s string, err error) {
    if err == nil {
        return
    }
    fmt.Println(s, err.Error())
    os.Exit(1)
}

//k8s源码摘抄
//https://github.com/kubernetes/Kubernetes/blob/v1.7.0-alpha.4/pkg/api/helper/helpers.go#L366
func ParseRFC3339(s string, nowFn func() metav1.Time) (metav1.Time, error) {
    if t, timeErr := time.Parse(time.RFC3339Nano, s); timeErr == nil {
        return metav1.Time{Time: t}, nil
    }
    t, err := time.Parse(time.RFC3339, s)
    if err != nil {
        return metav1.Time{}, err
    }
    return metav1.Time{Time: t}, nil
}
func TimeNow() metav1.Time {
    return metav1.Time{time.Now()}
}

//logOption 默认设置
func SetlogOption() {
    var err error
    //正则匹配pod名称
    if *n == "" {
        CheckError("请通过-n podname指定pod", errors.New(""))
    }

    if *ns == "" {
        *ns = "default"
    }

    //为了方便自己测试添加的
    var configfile string
    //configfile = "D:\\config"
    if len(*kubeconfig) != 0 {
        _, err = os.Stat(*kubeconfig)
        CheckError("指定的kubeconfig不存在", err)
        *kubeconfig = configfile
    } else if len(configfile) != 0 {
        *kubeconfig = configfile
    } else {
        home := homedir.HomeDir()
        if home != "" {
            *kubeconfig = filepath.Join(home, ".kube", "config")
        }
    }

    //设置日志参数
    logOptions.Follow = *follow   //是否开启持续输出
    logOptions.Timestamps = *timestamps //是否打开时间戳

    //时间相关参数设置
    var sinceSeconds time.Duration
    //同时设置就会报错
    if len(*sincetime) > 0 && len(*since) > 0 {
        CheckError("只能指定-s,-st其中之一", errors.New(""))
    }
    //如果都没有设置,就设置默认5分钟
    if len(*sincetime) == 0 && len(*since) == 0 {
        sinceSeconds, err = time.ParseDuration("5m")
        CheckError("since默认解析5m出错", err)
    }

    //如果since设置,则解析
    if len(*since) != 0 {
        sinceSeconds, err = time.ParseDuration(*since)
        CheckError("since指定时间格式不正确", err)
    }
    //如果sincetime设置,则默认+8小时
    if len(*sincetime) != 0 {
        if !strings.Contains(*sincetime, "+") {
            *sincetime = *sincetime + "+08:00"
        }
    }
    if sinceSeconds > 0 {
        // round up to the nearest second
        sec := int64(math.Ceil(float64(sinceSeconds) / float64(time.Second)))
        logOptions.SinceSeconds = &sec
    }
    //解析起始时间
    if ts, err := ParseRFC3339(*sincetime, TimeNow); err == nil {
        logOptions.SinceTime = &ts
    }

    //赋值获取行数,默认是全部
    if *tail > 0 {
        logOptions.TailLines = tail
    }

}

//发送信息到企业微信webhook+开机缓起作用
func SendMessage(url, msg string) {
    var m MessageMD
    m.MsgType = "markdown"
    m.Markdown.Content = msg
    jsons, err := json.Marshal(m)
    if err != nil {
        log.Println(err)
        return
    }
    resp := string(jsons)
    client := &http.Client{
        Timeout: 10 * time.Second,
    }
    req, err := http.NewRequest("POST", url, strings.NewReader(resp))
    if err != nil {
        log.Println(err)
        return
    }
    req.Header.Set("Content-Type", "application/json")
    r, err := client.Do(req)
    if err != nil {
        log.Println(err)
        return
    }
    defer r.Body.Close()
    _, err = ioutil.ReadAll(r.Body)
    if err != nil {
        log.Println(err)
        return
    }

}
举报

相关推荐

0 条评论