Bagaimana untuk melaksanakan logstash dalam golang

Mengenai pengurusan log, Logstash ialah alat popular yang mampu memproses, mengubah dan menghantar fail log dalam masa nyata. Walau bagaimanapun, dengan pembangunan berterusan seni bina perisian moden, semakin sukar bagi Logstash untuk memenuhi keperluan pemprosesan dan penyimpanan data yang kompleks. Untuk tujuan ini, bahasa Golang menyediakan pelaksanaan yang ringan dan cekap yang boleh disepadukan dengan mudah ke dalam pelbagai aliran kerja.

Artikel ini akan memperkenalkan cara menggunakan Golang untuk melaksanakan beberapa fungsi teras Logstash, termasuk membaca fail log, menghuraikan, menapis dan mengeluarkan ke lokasi sasaran. Kami juga akan membincangkan cara menggunakan alat penyimpanan dan pemindahan data biasa seperti ElasticSearch dan Kafka di Golang.

1. Pembacaan Fail

Sumber input yang paling biasa digunakan untuk Logstash ialah fail terlebih dahulu kita perlu menulis kod untuk membaca kandungan fail. Di Golang, Pengimbas yang paling biasa digunakan dalam pakej bufio boleh membaca fail baris demi baris dengan cekap.

file, err := os.Open("logfile.log")
if err != nil {
    // Handle error
}

scanner := bufio.NewScanner(file)

for scanner.Scan() {
    line := scanner.Text()
    // Process line
}

if err := scanner.Err(); err != nil {
    // Handle error
}

file.Close()

2. Analisis log

Logstash boleh menghuraikan fail log mengikut format yang berbeza, seperti JSON, XML, CSV, log Apache, dsb. Di Golang, tugasan ini boleh dicapai menggunakan pakej pengekodan/json, pengekodan/xml dan pengekodan/csv daripada pustaka standard. Ambil penghuraian data log dalam format JSON sebagai contoh:

type LogEntry struct {
    Timestamp string `json:"timestamp"`
    Message   string `json:"message"`
}

func parseJSON(line string) (*LogEntry, error) {
    entry := &LogEntry{}
    err := json.Unmarshal([]byte(line), entry)
    if err != nil {
        return nil, err
    }
    return entry, nil
}

3 Penapisan data

Satu lagi fungsi berkuasa Logstash ialah keupayaan untuk menapis dan mengubah suai data log, seperti memadam Medan yang tidak diperlukan. , menambah medan tambahan, medan pemformatan, dsb. Di Golang, anda boleh menggunakan struktur dan fungsi untuk melaksanakan logik pemprosesan ini. Sebagai contoh, kita boleh menyimpan dan mengendalikan data log dengan mentakrifkan struktur:

type LogEntry struct {
    Timestamp string `json:"timestamp"`
    Message   string `json:"message"`
}

type FilterConfig struct {
    RemoveFields []string `json:"remove_fields"`
    AddFields    map[string]interface{} `json:"add_fields"`
    DateFormat  string  `json:"date_format,omitempty"`
}

func applyFilter(config *FilterConfig, entry *LogEntry) {
    for _, field := range config.RemoveFields {
        delete(entry, field)
    }

    for key, value := range config.AddFields {
        entry[key] = value
    }

    if config.DateFormat != "" {
        // Convert timestamp to desired format
        // using format string
    }
}

4. Pemprosesan output

Logstash boleh mengeluarkan data log ke pelbagai lokasi sasaran, kaedah biasa Termasuk output ke ElasticSearch , Kafka, Redis, S3, dsb. Kami boleh menggunakan perpustakaan berkaitan di Golang untuk melaksanakan operasi ini. Sebagai contoh, output kepada ElasticSearch:

import (
    "context"
    "github.com/elastic/go-elasticsearch/v8"
    "github.com/elastic/go-elasticsearch/v8/esapi"
)

type ESOutputConfig struct {
    IndexName string `json:"index_name"`
    BatchSize int    `json:"batch_size"`
}

func createESOutput(config *ESOutputConfig) (*ElasticSearchOutput, error) {
    client, err := elasticsearch.NewDefaultClient()
    if err != nil {
        return nil, err
    }

    return &ElasticSearchOutput{
        client:    client,
        indexName: config.IndexName,
        batchSize: config.BatchSize,
    }, nil
}

func (out *ElasticSearchOutput) Write(entry *LogEntry) error {
    req := esapi.IndexRequest{
        Index:      out.indexName,
        DocumentID: "",
        Body:       strings.NewReader(entry.Message),
        Refresh:    "true",
    }

    res, err := req.Do(context.Background(), out.client)
    if err != nil {
        return err
    }

    defer res.Body.Close()

    if res.IsError() {
        return fmt.Errorf("failed to index log: %s", res.String())
    }
    return nil
}

5. Integrasikan ElasticSearch dan Kafka

Logstash Salah satu alat penyimpanan dan penghantaran data yang paling banyak digunakan ialah ElasticSearch dan Kafka. Di Golang, anda boleh menggunakan perpustakaan berkaitan untuk berinteraksi dengan perkhidmatan ini, seperti pakej go-elasticsearch ElasticSearch dan pakej sarama Kafka. Berikut ialah contoh penggunaan perpustakaan ini:

import (
    "github.com/Shopify/sarama"
    "github.com/elastic/go-elasticsearch/v8"
)

func main() {
    // Create ElasticSearch client
    esClient, _ := elasticsearch.NewDefaultClient()

    // Create Kafka producer
    kafkaConfig := sarama.NewConfig()
    producer, _ := sarama.NewAsyncProducer([]string{"localhost:9092"}, kafkaConfig)

    // Read log file
    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        line := scanner.Text()

        // Parse log entry from JSON
        entry, _ := parseJSON(line)

        // Apply filters
        applyFilter(config, entry)

        // Write to ElasticSearch
        createESOutput(config).Write(entry)

        // Write to Kafka
        KafkaOutput(producer, "my_topic").Write(entry)
    }
}

6. Ringkasan

Artikel ini memperkenalkan cara menggunakan Golang untuk melaksanakan fungsi teras Logstash, termasuk membaca fail log, menghurai, penapisan dan output ke lokasi sasaran. Kami juga membincangkan cara menggunakan alat storan dan pemindahan data biasa seperti ElasticSearch dan Kafka dengan Golang. Melalui alatan ini, kami boleh melaksanakan proses pengurusan log yang cekap, fleksibel dan boleh disesuaikan dengan mudah.

Atas ialah kandungan terperinci Bagaimana untuk melaksanakan logstash dalam golang. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!