k8s與日志--采用golang實 現Fluent Bit的output插件

s" "time" "unsafe" "github.com/Shopify/sarama" "github.com/fluent/fluent-bit-go/output" "github.com/ugorji/go/codec"
)

var (
 brokers []string
 producer sarama.SyncProducer
 timeout = 0 * time.Minute
 topic string
 module string
 messageKey string
)

//export FLBPluginRegister func FLBPluginRegister(ctx unsafe.Pointer) int {
 return output.FLBPluginRegister(ctx, "out_kafka", "Kafka Output Plugin.!")
}

//export FLBPluginInit // ctx (context) pointer to fluentbit context (state/ c code) func FLBPluginInit(ctx unsafe.Pointer) int {

 if bs := output.FLBPluginConfigKey(ctx, "brokers"); bs != "" {
 brokers = strings.Split(bs, ",")
 } else {
 log.Printf("you must set brokers")
 return output.FLB_ERROR
 }

 if tp := output.FLBPluginConfigKey(ctx, "topics"); tp != "" {
 topic = tp
 } else {
 log.Printf("you must set topics")
 return output.FLB_ERROR
 }

 if mo := output.FLBPluginConfigKey(ctx, "module"); mo != "" {
 module = mo
 } else {
 log.Printf("you must set module")
 return output.FLB_ERROR
 }

 if key := output.FLBPluginConfigKey(ctx, "message_key"); key != "" {
 messageKey = key
 } else {
 log.Printf("you must set message_key")
 return output.FLB_ERROR
 }

 config := sarama.NewConfig()
 config.Producer.Return.Successes = true if required_acks := output.FLBPluginConfigKey(ctx, "required_acks"); required_acks != "" {
 if acks, err := strconv.Atoi(required_acks); err == nil {
 config.Producer.RequiredAcks = sarama.RequiredAcks(acks)
 }
 }

 if compression_codec := output.FLBPluginConfigKey(ctx, "compression_codec"); compression_codec != "" {
 if codec, err := strconv.Atoi(compression_codec); err == nil {
 config.Producer.Compression = sarama.CompressionCodec(codec)
 }
 }

 if max_retry := output.FLBPluginConfigKey(ctx, "max_retry"); max_retry != "" {
 if max_retry, err := strconv.Atoi(max_retry); err == nil {
 config.Producer.Retry.Max = max_retry
 }
 }

 if timeout == 0 {
 timeout = 5 * time.Minute
 }
 // If Kafka is not running on init, wait to connect
 deadline := time.Now().Add(timeout)
 for tries := 0; time.Now().Before(deadline); tries++ {
 var err error
 if producer == nil {
 producer, err = sarama.NewSyncProducer(brokers, config)
 }
 if err == nil {
 return output.FLB_OK
 }
 log.Printf("Cannot connect to Kafka: (%s) retrying...", err)
 time.Sleep(time.Second * 30)
 }

 log.Printf("Kafka failed to respond after %s", timeout)
 return output.FLB_ERROR

}

//export FLBPluginFlush // FLBPluginFlush is called from fluent-bit when data need to be sent. is called from fluent-bit when data need to be sent. func FLBPluginFlush(data unsafe.Pointer, length C.int, tag *C.char) int {
 var h codec.MsgpackHandle

 var b []byte var m interface{}
 var err error

 b = C.GoBytes(data, length)
 dec := codec.NewDecoderBytes(b, &h)

 // Iterate the original MessagePack array var msgs []*sarama.ProducerMessage
 for {
 // decode the msgpack data
 err = dec.Decode(&m)
 if err != nil {
 if err == io.EOF {
 break
 }
 log.Printf("Failed to decode msgpack data: %v\n", err)
 return output.FLB_ERROR
 }

 // Get a slice and their two entries: timestamp and map
 slice := reflect.ValueOf(m)
 data := slice.Index(1)

 // Convert slice data to a real map and iterate
 mapData := data.Interface().(map[interface{}]interface{})
 flattenData, err := Flatten(mapData, "", UnderscoreStyle)
 if err != nil {
 break
 }

 message := ""
 host := "" for k, v := range flattenData {
 value := "" switch t := v.(type) {
 case string:
 value = t
 case []byte:
 value = string(t)
 default:
 value = fmt.Sprintf("%v", v)
 }

 if k == "pod_name" {
 host = value
 }

 if k == messageKey {
 message = value
 }

 }
 if message == "" || host == "" {
 break
 }

 m := &sarama.ProducerMessage{
 Topic: topic,
 Key: sarama.StringEncoder(fmt.Sprintf("host=%s|module=%s", host, module)),
 Value: sarama.ByteEncoder(message),
 }
 msgs = append(msgs, m)

 }

 err = producer.SendMessages(msgs)
 if err != nil {
 log.Printf("FAILED to send kafka message: %s\n", err)
 return output.FLB_ERROR
 }
 return output.FLB_OK
}

//export FLBPluginExit func FLBPluginExit() int {
 producer.Close()
 return output.FLB_OK
}

func main() {
}           

• FLBPluginExit 插件退出的時候需要執行的一些方法，比如關閉連接配接。
• FLBPluginRegister 注冊插件
• FLBPluginInit 插件初始化
• FLBPluginFlush flush到資料到output
• FLBPluginConfigKey 擷取配置檔案中參數

PS

當然除了FLBPluginConfigKey之外，也可以通過擷取環境變量來獲得設定參數。

ctx相當于一個上下文，負責之間的資料的傳遞。

編譯和執行

編譯的時候

go build -buildmode=c-shared -o out_kafka.so .           

生成out_kafka.so

執行的時候

/fluent-bit/bin/fluent-bit" -c /fluent-bit/etc/fluent-bit.conf -e /fluent-bit/out_kafka.so           

總結

采用類似的編寫結構，就可以定制化自己的輸出插件了。

本文轉自SegmentFault-

k8s與日志--采用golang實作Fluent Bit的output插件