コミュニティ 学ぶ ツールライブラリ レジャー
検索
日本語
ホームページ > バックエンド開発 > Golang > 本文

Go言語で実装されたログ収集システムの詳細なグラフィックとテキストの説明

尚
リリース： 2019-11-27 15:47:35
転載
4067 人が閲覧しました

Go言語で実装されたログ収集システムの詳細なグラフィックとテキストの説明

#このログ収集システムの枠組みを以下のように整理しました。

今回実装するコードの全体的なロジックは次のとおりです。 Go言語で実装されたログ収集システムの詳細なグラフィックとテキストの説明

##完全なコードのアドレスは次のとおりです: https://github.com/pythonsite/logagent

Go言語で実装されたログ収集システムの詳細なグラフィックとテキストの説明

etcd の概要 #可用性の高い分散型 Key-Value ストレージは、構成共有とサービス検出に使用できます

類似プロジェクト: Zookeeper と Consul

開発言語: go

インターフェース: 安静なインターフェースを提供しますシンプルな

実装アルゴリズム: raft アルゴリズムに基づく強力な一貫性、高可用性サービス ストレージ ディレクトリ

etcd のアプリケーション シナリオ:

1. サービスの検出とサービスの登録

2. 構成センター (実装したログ収集クライアントによって使用する必要があります)

#3. 分散ロック

4. マスター選挙

公式 Web サイトetcd 非常に簡潔な紹介があります:

##etcd ビルド:

ダウンロード アドレス: https://github.com/coreos/etcd/releases/

ベース自分の環境で対応するバージョンをダウンロードして起動するだけです。Go言語で実装されたログ収集システムの詳細なグラフィックとテキストの説明

起動後、次のコマンドで確認できます:

[root@localhost etcd-v3.2.18-linux-amd64]# ./etcdctl set name zhaofan 

zhaofan
[root@localhost etcd-v3.2.18-linux-amd64]# ./etcdctl get name
zhaofan
[root@localhost etcd-v3.2.18-linux-amd64]#
ログイン後にコピー


context 導入と使用

実は、これはコンテキスト管理と訳されるのですが、コンテキストの役割とは何でしょうか?主に次の 2 つの機能があります: 1. goroutine のタイムアウトを制御する

2 、コンテキスト データの保存

次の簡単な例を通じて理解します: 

package main

import (
    "fmt"
    "time"
    "net/http"
    "context"
    "io/ioutil"
)


type Result struct{
    r *http.Response
    err error
}

func process(){
    ctx,cancel := context.WithTimeout(context.Background(),2*time.Second)
    defer cancel()
    tr := &http.Transport{}
    client := &http.Client{Transport:tr}
    c := make(chan Result,1)
    req,err := http.NewRequest("GET","http://www.google.com",nil)
    if err != nil{
        fmt.Println("http request failed,err:",err)
        return
    }
    // 如果请求成功了会将数据存入到管道中
    go func(){
        resp,err := client.Do(req)
        pack := Result{resp,err}
        c <- pack
    }()

    select{
    case <- ctx.Done():
        tr.CancelRequest(req)
        fmt.Println("timeout!")
    case res := <-c:
        defer res.r.Body.Close()
        out,_:= ioutil.ReadAll(res.r.Body)
        fmt.Printf("server response:%s",out)
    }
    return

}

func main() {
    process()
}
ログイン後にコピー

コンテキストを介してコンテキストを保存するコンテキストを記述します。コード例は次のとおりです: 

package main

import (
    "github.com/Go-zh/net/context"
    "fmt"
)

func add(ctx context.Context,a,b int) int {
    traceId := ctx.Value("trace_id").(string)
    fmt.Printf("trace_id:%v\n",traceId)
    return a+b
}

func calc(ctx context.Context,a, b int) int{
    traceId := ctx.Value("trace_id").(string)
    fmt.Printf("trace_id:%v\n",traceId)
    //再将ctx传入到add中
    return add(ctx,a,b)
}

func main() {
    //将ctx传递到calc中
    ctx := context.WithValue(context.Background(),"trace_id","123456")
    calc(ctx,20,30)

}
ログイン後にコピー

etcd と context の結合

Go を介して etcd を接続する簡単な例: (注意が必要な小さな問題は、etcd の起動方法です。デフォルトの起動では、特に仮想的にインストールされている場合は接続できない可能性があるため、次のコマンドで起動する必要があります: ./etcd --listen-client-urls http://0.0.0.0:2371 --advertise -client-urls http://0.0.0.0: 2371 --listen-peer-urls http://0.0.0.0:2381)

package main

import (
    etcd_client "github.com/coreos/etcd/clientv3"
    "time"
    "fmt"
)

func main() {
    cli, err := etcd_client.New(etcd_client.Config{
        Endpoints:[]string{"192.168.0.118:2371"},
        DialTimeout:5*time.Second,
    })
    if err != nil{
        fmt.Println("connect failed,err:",err)
        return
    }

    fmt.Println("connect success")
    defer cli.Close()
}
ログイン後にコピー

次の例は、etcd に接続し、保存して値を取得します

package main

import (
    "github.com/coreos/etcd/clientv3"
    "time"
    "fmt"
    "context"
)

func main() {
    cli,err := clientv3.New(clientv3.Config{
        Endpoints:[]string{"192.168.0.118:2371"},
        DialTimeout:5*time.Second,
    })
    if err != nil{
        fmt.Println("connect failed,err:",err)
        return
    }
    fmt.Println("connect succ")
    defer cli.Close()
    ctx,cancel := context.WithTimeout(context.Background(),time.Second)
    _,err = cli.Put(ctx,"logagent/conf/","sample_value")
    cancel()
    if err != nil{
        fmt.Println("put failed,err",err)
        return
    }
    ctx, cancel = context.WithTimeout(context.Background(),time.Second)
    resp,err := cli.Get(ctx,"logagent/conf/")
    cancel()
    if err != nil{
        fmt.Println("get failed,err:",err)
        return
    }
    for _,ev := range resp.Kvs{
        fmt.Printf("%s:%s\n",ev.Key,ev.Value)
    }
}
ログイン後にコピー

コンテキストについて公式 Web サイトには、開いたゴルーチンの終了を制御するために使用される非常に便利な例もあります。コードは次のとおりです。公式 Web サイトのドキュメントの WithDeadline デモの例: 
package main

import (
    "context"
    "fmt"
)

func main() {
    // gen generates integers in a separate goroutine and
    // sends them to the returned channel.
    // The callers of gen need to cancel the context once
    // they are done consuming generated integers not to leak
    // the internal goroutine started by gen.
    gen := func(ctx context.Context) <-chan int {
        dst := make(chan int)
        n := 1
        go func() {
            for {
                select {
                case <-ctx.Done():
                    return // returning not to leak the goroutine
                case dst <- n:
                    n++
                }
            }
        }()
        return dst
    }

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel() // cancel when we are finished consuming integers

    for n := range gen(ctx) {
        fmt.Println(n)
        if n == 5 {
            break
        }
    }
}
ログイン後にコピー

上記のコードを通じて、基本的な使用法を示します。次に、構成管理に etcd を使用する場合、構成が変更された場合、対応するサーバーにどのように通知しますか

package main


import (
    "context"
    "fmt"
    "time"
)

func main() {
    d := time.Now().Add(50 * time.Millisecond)
    ctx, cancel := context.WithDeadline(context.Background(), d)

    // Even though ctx will be expired, it is good practice to call its
    // cancelation function in any case. Failure to do so may keep the
    // context and its parent alive longer than necessary.
    defer cancel()

    select {
    case <-time.After(1 * time.Second):
        fmt.Println("overslept")
    case <-ctx.Done():
        fmt.Println(ctx.Err())
    }

}
ログイン後にコピー

Kafka コンシューマ コードの単純な実装 例: 

package main

import (
    "github.com/coreos/etcd/clientv3"
    "time"
    "fmt"
    "context"
)

func main() {
    cli,err := clientv3.New(clientv3.Config{
        Endpoints:[]string{"192.168.0.118:2371"},
        DialTimeout:5*time.Second,
    })
    if err != nil {
        fmt.Println("connect failed,err:",err)
        return
    }
    defer cli.Close()
    // 这里会阻塞
    rch := cli.Watch(context.Background(),"logagent/conf/")
    for wresp := range rch{
        for _,ev := range wresp.Events{
            fmt.Printf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
        }
    }
}
ログイン後にコピー

ただし、上記のコードは最適なコードではありません。 time.sleep による goroutine の実行を変更して、sync.WaitGroup

package main

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

func main() {
    consumer,err := sarama.NewConsumer(strings.Split("192.168.0.118:9092",","),nil)
    if err != nil{
        fmt.Println("failed to start consumer:",err)
        return
    }
    partitionList,err := consumer.Partitions("nginx_log")
    if err != nil {
        fmt.Println("Failed to get the list of partitions:",err)
        return
    }
    fmt.Println(partitionList)
    for partition := range partitionList{
        pc,err := consumer.ConsumePartition("nginx_log",int32(partition),sarama.OffsetNewest)
        if err != nil {
            fmt.Printf("failed to start consumer for partition %d:%s\n",partition,err)
            return
        }
        defer pc.AsyncClose()
        go func(partitionConsumer sarama.PartitionConsumer){
            for msg := range pc.Messages(){
                fmt.Printf("partition:%d Offset:%d Key:%s Value:%s",msg.Partition,msg.Offset,string(msg.Key),string(msg.Value))
            }
        }(pc)
    }
    time.Sleep(time.Hour)
    consumer.Close()

}
ログイン後にコピー

クライアントが収集する必要があるログ情報を etcd

に配置します。

etcd 処理のコードは次のとおりです: 

package main

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

var (
    wg sync.WaitGroup
)

func main() {
    consumer,err := sarama.NewConsumer(strings.Split("192.168.0.118:9092",","),nil)
    if err != nil{
        fmt.Println("failed to start consumer:",err)
        return
    }
    partitionList,err := consumer.Partitions("nginx_log")
    if err != nil {
        fmt.Println("Failed to get the list of partitions:",err)
        return
    }
    fmt.Println(partitionList)
    for partition := range partitionList{
        pc,err := consumer.ConsumePartition("nginx_log",int32(partition),sarama.OffsetNewest)
        if err != nil {
            fmt.Printf("failed to start consumer for partition %d:%s\n",partition,err)
            return
        }
        defer pc.AsyncClose()
        go func(partitionConsumer sarama.PartitionConsumer){
            wg.Add(1)
            for msg := range partitionConsumer.Messages(){
                fmt.Printf("partition:%d Offset:%d Key:%s Value:%s",msg.Partition,msg.Offset,string(msg.Key),string(msg.Value))
            }
            wg.Done()
        }(pc)
    }

    //time.Sleep(time.Hour)
    wg.Wait()
    consumer.Close()

}
ログイン後にコピー
同様に、ここに適切な制限を追加します。結局のところ、ログ収集プログラムは現在のビジネスのパフォーマンスに影響を与えることができないため、limit.go が速度を制限する: 
package main

import (
    "github.com/coreos/etcd/clientv3"
    "time"
    "github.com/astaxie/beego/logs"
    "context"
    "fmt"
)

var Client *clientv3.Client
var logConfChan chan string


// 初始化etcd
func initEtcd(addr []string,keyfmt string,timeout time.Duration)(err error){

    var keys []string
    for _,ip := range ipArrays{
        //keyfmt = /logagent/%s/log_config
        keys = append(keys,fmt.Sprintf(keyfmt,ip))
    }

    logConfChan = make(chan string,10)
    logs.Debug("etcd watch key:%v timeout:%v", keys, timeout)

    Client,err = clientv3.New(clientv3.Config{
        Endpoints:addr,
        DialTimeout: timeout,
    })
    if err != nil{
        logs.Error("connect failed,err:%v",err)
        return
    }
    logs.Debug("init etcd success")
    waitGroup.Add(1)
    for _, key := range keys{
        ctx,cancel := context.WithTimeout(context.Background(),2*time.Second)
        // 从etcd中获取要收集日志的信息
        resp,err := Client.Get(ctx,key)
        cancel()
        if err != nil {
            logs.Warn("get key %s failed,err:%v",key,err)
            continue
        }

        for _, ev := range resp.Kvs{
            logs.Debug("%q : %q\n",  ev.Key, ev.Value)
            logConfChan <- string(ev.Value)
        }
    }
    go WatchEtcd(keys)
    return
}

func WatchEtcd(keys []string){
    // 这里用于检测当需要收集的日志信息更改时及时更新
    var watchChans []clientv3.WatchChan
    for _,key := range keys{
        rch := Client.Watch(context.Background(),key)
        watchChans = append(watchChans,rch)
    }

    for {
        for _,watchC := range watchChans{
            select{
            case wresp := <-watchC:
                for _,ev:= range wresp.Events{
                    logs.Debug("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
                    logConfChan <- string(ev.Kv.Value)
                }
            default:

            }
        }
        time.Sleep(time.Second)
    }
    waitGroup.Done()
}

func GetLogConf()chan string{
    return logConfChan
}
ログイン後にコピー
推奨: 言語チュートリアルに進む

以上がGo言語で実装されたログ収集システムの詳細なグラフィックとテキストの説明の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

関連ラベル：
言語を移動
ソース：cnblogs.com
前の記事：Go言語タイムパッケージのいくつかの使用方法 次の記事：Go言語での例外処理の紹介
このウェブサイトの声明
この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。
著者別の最新記事
最新の問題
MySQL: 「<ここの変数またはサブクエリ> として...からテキストを選択します」 次のデータを含む次のテーブルがあります: id text language 1 german text german 2 english text english 私が望むのは、次...
から 2024-04-06 19:25:54
0
2
529
透過的なブラウザを起動して、Web アプリケーションと PWA のエクスペリエンスを向上させます [静的 HTML、CSS、JavaScript をサポート] [ユーザーが Web アプリケーション/PWA でデスクトップの壁紙を表示できるようにします] 私は現在、ソケットを使用した Web アプリケーション (チャット) を開発しているジュニアのフロントエンド開発者兼デザイナーです。私と友人が使用している言語は、Scala、HTM...
から 2024-04-05 13:50:12
0
1
1367
正規表現を使用した HTML での単語カウント これもこれと同じ問題です。しかし、私はJavaScriptを使用していないので、「innerText」は私にとって解決策ではなく、正規表現が一連の処理を行わずに/(<.*?&...
から 2024-04-01 22:44:06
0
1
416
ユーザーがあるページで画像をアップロードまたは変更し、別のページに表示できるようにするにはどうすればよいですか? ユーザーがあるページに写真をアップロードして、別のページに表示できるようにするにはどうすればよいですか?ユーザーは画像を変更でき、他のページでも同様に変更されます。また、php/j...
から 2024-04-01 18:33:31
0
1
332
Python: Pymysql が応答なしでサイレントにハングしますか? Python を使用した最初のアプリケーション。アプリケーションが pymysql 呼び出しでハングします。このライブラリは安定していますか?ここで MySQL データベースに接続...
から 2024-04-01 10:55:37
0
1
321
関連トピック
詳細>
人気のおすすめ
人気のチュートリアル
詳細>
関連するチュートリアル
人気のおすすめ
最新のコース
最新のダウンロード
詳細>
ウェブエフェクト
公式サイト
サイト素材
フロントエンドテンプレート