forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
9 changed files
with
676 additions
and
173 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,7 +1,8 @@ | ||
| github.com/mattbaird/elastigo/api | ||
| github.com/mattbaird/elastigo/core | ||
| github.com/packetbeat/elastigo/api | ||
| github.com/packetbeat/elastigo/core | ||
| github.com/nranchev/go-libGeoIP | ||
| github.com/BurntSushi/toml | ||
| github.com/akrennmair/gopcap | ||
| github.com/tsg/fsnotify | ||
| github.com/garyburd/redigo/redis | ||
| labix.org/v2/mgo/bson |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,152 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "fmt" | ||
| "strings" | ||
| "github.com/packetbeat/elastigo/api" | ||
| "github.com/packetbeat/elastigo/core" | ||
| ) | ||
|
|
||
| type ElasticsearchOutputType struct { | ||
| OutputInterface | ||
| Index string | ||
| TopologyExpire int | ||
|
|
||
| TopologyMap map[string]string | ||
| } | ||
|
|
||
| type PublishedTopology struct { | ||
| Name string | ||
| IPs string | ||
| } | ||
|
|
||
| var ElasticsearchOutput ElasticsearchOutputType | ||
|
|
||
| func (out *ElasticsearchOutputType) Init(config tomlMothership) error { | ||
|
|
||
| api.Domain = config.Host | ||
| api.Port = fmt.Sprintf("%d", config.Port) | ||
| api.Username = config.Username | ||
| api.Password = config.Password | ||
| api.BasePath = config.Path | ||
|
|
||
| if config.Protocol != "" { | ||
| api.Protocol = config.Protocol | ||
| } | ||
|
|
||
| if config.Index != "" { | ||
| out.Index = config.Index | ||
| } else { | ||
| out.Index = "packetbeat" | ||
| } | ||
|
|
||
| out.TopologyExpire = 15000 | ||
| if _Config.Agent.Topology_expire != 0 { | ||
| out.TopologyExpire = _Config.Agent.Topology_expire /*sec*/ * 1000 // millisec | ||
| } | ||
|
|
||
| err := out.EnableTTL() | ||
| if err != nil { | ||
| ERR("Fail to set _ttl mapping: %s", err) | ||
| return err | ||
| } | ||
|
|
||
| INFO("[ElasticsearchOutput] Using Elasticsearch %s://%s:%s%s", api.Protocol, api.Domain, api.Port, api.BasePath) | ||
| INFO("[ElasticsearchOutput] Using index pattern [%s-]YYYY.MM.DD", out.Index) | ||
| INFO("[ElasticsearchOutput] Topology expires after %ds", out.TopologyExpire / 1000) | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (out *ElasticsearchOutputType) EnableTTL() error { | ||
| setting := map[string]interface{}{ | ||
| "server-ip": map[string]interface{}{ | ||
| "_ttl": map[string]string{"enabled": "true", "default": "15000"}, | ||
| }, | ||
| } | ||
|
|
||
| // Make sure the index exists | ||
| _, err := core.Index("packetbeat-topology", "", "", nil, nil) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| _, err = core.Index("packetbeat-topology", "server-ip", "_mapping", nil, setting) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (out *ElasticsearchOutputType) GetNameByIP(ip string) string { | ||
| name, exists := out.TopologyMap[ip] | ||
| if !exists { | ||
| return "" | ||
| } | ||
| return name | ||
| } | ||
| func (out *ElasticsearchOutputType) PublishIPs(name string, localAddrs []string) error { | ||
| DEBUG("output_elasticsearch", "Publish IPs %s with expiration time %d", localAddrs, out.TopologyExpire) | ||
| _, err := core.IndexWithParameters( | ||
| "packetbeat-topology", /*index*/ | ||
| "server-ip", /*type*/ | ||
| name, /* id */ | ||
| "", /*parent id */ | ||
| 0, /* version */ | ||
| "", /* op_type */ | ||
| "", /* routing */ | ||
| "", /* timestamp */ | ||
| out.TopologyExpire, /*ttl*/ | ||
| "", /* percolate */ | ||
| "", /* timeout */ | ||
| false, /*refresh */ | ||
| nil, /*args */ | ||
| PublishedTopology{name, strings.Join(localAddrs, ",")} /* data */) | ||
|
|
||
| if err != nil { | ||
| ERR("Fail to publish IP addresses: %s", err) | ||
| return err | ||
| } | ||
|
|
||
| out.UpdateLocalTopologyMap() | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (out *ElasticsearchOutputType) UpdateLocalTopologyMap() { | ||
|
|
||
| // get all agents IPs from Elasticsearch | ||
| TopologyMapTmp := make(map[string]string) | ||
|
|
||
| res, err := core.SearchUri("packetbeat-topology", "server-ip", nil) | ||
| if err == nil { | ||
| for _, server := range res.Hits.Hits { | ||
| var pub PublishedTopology | ||
| err = json.Unmarshal([]byte(*server.Source), &pub) | ||
| if err != nil { | ||
| ERR("json.Unmarshal fails with: %s", err) | ||
| } | ||
| // add mapping | ||
| ipaddrs := strings.Split(pub.IPs, ",") | ||
| for _, addr := range ipaddrs { | ||
| TopologyMapTmp[addr] = pub.Name | ||
| } | ||
| } | ||
| } else { | ||
| ERR("Getting topology map fails with: %s", err) | ||
| } | ||
|
|
||
| // update topology map | ||
| out.TopologyMap = TopologyMapTmp | ||
|
|
||
| DEBUG("output_elasticsearch", "Topology map %s", out.TopologyMap) | ||
| } | ||
|
|
||
| func (out *ElasticsearchOutputType) PublishEvent(event *Event) error { | ||
|
|
||
| index := fmt.Sprintf("%s-%d.%02d.%02d", out.Index, event.Timestamp.Year(), event.Timestamp.Month(), event.Timestamp.Day()) | ||
| _, err := core.Index(index, event.Type, "", nil, event) | ||
| DEBUG("output_elasticsearch", "Publish event") | ||
| return err | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.