gopipe下載 - gopipe原始碼下載

gopipe

其他類別

1.0.0

下載

走管

用Go編寫的處理管道（類似logstash）

快速啟動

克隆並安裝相依性：

 mkdir -p $GOPATH/src/github.com/urban-1
cd $GOPATH/src/github.com/urban-1
git clone [email protected]:urban-1/gopipe.git
dep ensure

您可以使用一個 Makefile 來快速建置並開始處理該專案：

 $ make help

Available targets:

  rdkafka         Build librdkafka locally
  gopipe          Build gopipe
  tests           Run tests (includes fmt and vet)
  show_coverage   Show coverage results

建築學

我們的目標是定義一個以模組化方式接收和處理事件的管道。事件可以包含：

原始未格式化位元組 ( Data["raw"] )
字串內容 ( Data["message"] )
解碼後的 JSON 值（ Data為map[string]interface{} ）

我們將我們的模組稱為“組件”，這些模組分為三類：

輸入：產生事件並將其推送到 Go 通道的元件
Proc：修改事件資料的處理組件。他們從 Go 通道讀取事件並將其寫入到 Go 通道
輸出：只有讀取事件和某種形式的輸出。他們通常不會將它們推回 Go 通道，而是丟棄它們。

本質上，所有元件都是「相同的」（實現相同的介面）。唯一的區別在於他們可以使用哪些管道。

為什麼和如何？

嗯...我在 C++ 中做了類似的事情來處理網路串流資料包，並且認為 Go（真的）很快並且並發是此類應用程式的完美匹配。
為什麼不使用現有的東西：我們可以擴展現有框架，但是，這是一個複製 C++ 程式碼的 Go 學習練習...
那有什麼不同呢？我們關注系統的角度，我們希望這個框架更加面向網路/資料而不是面向log ：
- 原始資料處理 - 請參閱flowreplicator.json配置。
- 允許使用者編寫更多邏輯，仍然基於配置（if/else 支援）
- （實驗）支援任務來饋送和更新處理模組的數據
未來的計劃是什麼：我們計劃維護和擴展它，直到我們完全移植我們的 C++ 程式碼...維護將繼續，但我們希望在社區的幫助下根據需要進行擴展。

成分

輸入

TCP ：支援原始、字串、CSV 和 JSON
UDP ：支援原始、字串、CSV 和 JSON
Kafka ：支援原始、字串、CSV 和 JSON

加工

新增字段：根據靜態值或表達式新增字段
新增時間：為資料新增時間戳
Cast ：將欄位轉換為不同的資料類型
刪除字段：從資料中刪除字段
If/Else ：使用 if/else/endif 控制流程
在列表中：根據值列表檢查字段
Log ：將事件資料記錄到標準輸出
最長前綴匹配：執行 LPM 並將元數據附加到事件數據
MD5 ：哈希事件的字段
正規表示式：將字串事件轉換為資料事件
採樣器：選擇性轉送事件（每 X 一個）

輸出

檔案：支援 CSV 和 JSON
Null ：黑洞事件
UDP ：支援原始和字串

配置範例

UDP流複製器

將複製並可選擇採樣 UDP 封包：

 {
    "main": {
        "num_cpus": 2,
        "log_level": 1,
        "channel_size": 50000,
        "stats_every": 100000
    },
    "in": {
        "module": "UDPRawInput",
        "listen": "0.0.0.0",
        "port": 9090
    },
    "proc": [
        {
            "module": "SamplerProc",
            "every": 2
        },
        {
            "module": "UDPRawOutput",
            "target": "127.0.0.1",
            "port": 9091
        }
    ],
    "out": {
        "module": "UDPRawOutput",
        "target": "127.0.0.1",
        "port": 9092
    }
}

設定日誌行格式 + if/else 演示

透過 TCP 接收行，將其解析為資料字段，添加時間戳，將某些資料轉換為不同的資料類型，丟棄原始訊息，散列某些字段等

 {
    "main": {
        "num_cpus": 2,
        "log_level": 1,
        "channel_size": 50000,
        "stats_every": 100000
    },
    "in": {
        "module": "TCPStrInput",
        "listen": "0.0.0.0",
        "port": 9092,
        "headers": ["hello", "test", "src"],
        "separator": ",",
        "convert": false
    },
    "proc": [
        {
            "module": "RegexProc",
            "regexes": [
                "(?mi)(?P<host>[0-9a-z]+) (?P<port>[0-9]+): (?P<hostEvent>.*)"
            ]
        },
        {
            "module": "DropFieldProc",
            "field_name": "message"
        },
        {
            "module": "CastProc",
            "fields": ["port"],
            "types": ["int"]
        },
        {
            "module": "InListProc",
            "in_field": "port",
            "out_field": "port_block",
            "reload_minutes": 100000000,
            "list": ["8080", "443", "23230", "14572", "17018"]
        },
        {"module": "if",  "condition": "port_block == true "},
            {
                "module": "Md5Proc",
                "in_fields": ["host"],
                "out_fields": ["host_hash"],
                "salt": "andPepper!"
            },
        {"module": "else"},
            {
                "module": "AddTimeProc",
                "field_name": "_timestamp"
            },
        {"module": "endif"},
        {
            "module": "LogProc",
            "level": "info"
        }
    ],
    "out": {
        "module": "FileJSONOutput",
        "rotate_seconds": 60,
        "folder": "/tmp",
        "file_name_format": "gopipe-20060102-150405.json"
    }
}

最長前綴匹配

在偵聽 JSON 行的 TCP 套接字上接收：

 {
    "main": {
        "num_cpus": 2,
        "log_level": 1,
        "channel_size": 50000,
        "stats_every": 10000000
    },
    "in": {
        "module": "TCPJSONInput",
        "listen": "0.0.0.0",
        "port": 9092
    },
    "proc": [
        {
            "module": "AddTimeProc",
            "field_name": "_timestamp"
        },
        {
            "module": "LPMProc",
            "filepath": "/tmp/prefix-asn.txt",
            "reload_minutes": 1440,
            "in_fields": ["src", "dst"],
            "out_fields": [
                {"newkey": "_{{in_field}}_prefix", "metakey": "prefix"},
                {"newkey": "_{{in_field}}_asn", "metakey": "asn"}
            ]
        }
    ],
    "out": {
        "module": "FileJSONOutput",
        "rotate_seconds": 60,
        "folder": "/tmp",
        "file_name_format": "gopipe-20060102-150405.json"
    }
}

任務

以下配置部分定義了每 10 秒運行一次的任務。通常您希望更新InListProc和LPMProc元件的檔案來源...在這種情況下，您的想法是您系統中的某個位置有一個小的 shell 腳本來更新本機檔案。然後您需要“調用”重新加載以將新數據加載到內存中：

 ...
    ],
    "out": {
        "module": "FileJSONOutput",
        "rotate_seconds": 60,
        "folder": "/tmp",
        "file_name_format": "gopipe-20060102-150405.json"
    },
    "tasks": [
        {
            "name": "LSing...",
            "command": ["ls", "-al"],
            "interval_seconds": 10,
            "signals": [
                {"mod": 4, "signal": "reload"}
            ]
        },
        {
            "name": "LSing...2",
            "command": ["ls", "-al"],
            "interval_seconds": 10,
            "signals": []
        }
    ]
...

上面我們定義了兩個任務。它們之間的區別在於，如果組件運行成功，第一個組件將向其發出信號。訊號"reload"將被傳送到元件4 ，並由該元件來處理它。

元件索引定義為此元件在 config 中的順序，包括輸入元件。鑑於目前我們只支援一個輸入，上面的組件4是proc部分的第三個。