日志收集系统

package main

import (
    "fmt"
    "github.com/Shopify/sarama"
)


func main() {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll  //发送消息给kafka，等待kafka返回ack，确认收到消息。防止消息丢失。
    config.Producer.PartitiOner= sarama.NewRandomPartitioner //分区的策略是随机的
    config.Producer.Return.Successes = true
    msg := &sarama.ProducerMessage{}
    msg.Topic = "nginx_log"
    msg.Value = sarama.StringEncoder("this is a good test, my message is good")  //存入的日志
    client, err := sarama.NewSyncProducer([]string{"192.168.31.177:9092"}, config) //配置kafka的IP，生成一个客户端
    if err != nil {
        fmt.Println("producer close, err:", err)
        return
    }
    defer client.Close()
    pid, offset, err := client.SendMessage(msg)  //pid分区号，offset偏移
    if err != nil {
        fmt.Println("send message failed,", err)
        return
    }
    fmt.Printf("pid:%v offset:%v\n", pid, offset)
}

package main


import (
    "fmt"
    "github.com/hpcloud/tail"
    "time"
)

func main() {
    filename := "./my.log"  //要读取的文件的名字
    tails, err := tail.TailFile(filename, tail.Config{
        ReOpen:    true,
        Follow:    true,
        Location: &tail.SeekInfo{Offset: 0, Whence: 2},  //2从末尾开始读
        MustExist: false, //文件不存在也不会报错
        Poll:    true, //轮询
    })

    if err != nil {
        fmt.Println("tail file err:", err)
        return
    }
    var msg *tail.Line
    var ok bool
    for true {
        msg, ok = <-tails.Lines
        if !ok {
            fmt.Printf("tail file close reopen, filename:%s\n", tails.Filename)
            time.Sleep(100 * time.Millisecond)
            continue
        }
    fmt.Println("msg:", msg.Text)    
    }
}

package main

import (
    "fmt"
    "github.com/astaxie/beego/config"
)


func main() {
    conf, err := config.NewConfig("ini", "./logcollect.conf")
    if err != nil {
        fmt.Println("new config failed, err:", err)
        return
    }
    port, err := conf.Int("server::port")
    if err != nil {
        fmt.Println("read server:port failed, err:", err)
        return
    }


    fmt.Println("Port:", port)
    log_level, err := conf.Int("log::log_level")
    if err != nil {
        fmt.Println("read log_level failed, ", err)
        return
    }

    fmt.Println("log_level:", log_level)
    log_path := conf.String("log::log_path")
    fmt.Println("log_path:", log_path)

}

package main

import (
    "encoding/json"
    "fmt"
    "github.com/astaxie/beego/logs"
)


func main() {
    config := make(map[string]interface{})
    config["filename"] = "./logs/logcollect.log"
    config["level"] = logs.LevelDebug
    configStr, err := json.Marshal(config)
    if err != nil {
        fmt.Println("marshal failed, err:", err)
        return
    }

    logs.SetLogger(logs.AdapterFile, string(configStr))
    logs.Debug("this is a test, my name is %s", "stu01")
    logs.Trace("this is a trace, my name is %s", "stu02")
    logs.Warn("this is a warn, my name is %s", "stu03")

}