HTTP http://e064eb4b0aa993db28ad513 ... :9200

HTTPS https://e064eb4b0aa993db28ad51 ... :9243

 curl -u elasticsearch-cn:elasticsearch-cn http://e064eb4b0aa993db28ad513 ... 9200/

{

  "name" : "instance-0000000009",

  "cluster_name" : "e064eb4b0aa993db28ad513e4d2df5e3",

  "version" : {

    "number" : "2.1.1",

    "build_hash" : "40e2c53a6b6c2972b3d13846e450e66f4375bd71",

    "build_timestamp" : "2015-12-15T13:05:55Z",

    "build_snapshot" : false,

    "lucene_version" : "5.3.1"

  },

  "tagline" : "You Know, for Search"

}

#创建相应目录

mkdir -p $GOPATH/src/github.com/elastic/ 

cd $GOPATH/src/github.com/elastic



#签出源码

git clone https://github.com/elastic/beats.git

cd beats



#修改官方仓库为upstream源，设置自己的仓库为origin源

git remote rename origin upstream

git remote add origin git@github.com:medcl/packetbeat.git



#获取上游最新的代码，如果是刚fork的话可不用管

git pull upstream master



#签出一个名为smtpbeat的分支，用于开发这个功能

git checkout -b smtpbeat



#切换到packetbeat模块

cd packetbeat



#获取依赖信息

(mkdir -p $GOPATH/src/golang.org/x/&&cd $GOPATH/src/golang.org/x &&git clone https://github.com/golang/tools.git )

go get github.com/tools/godep



#编译

make

output:

  elasticsearch:

    enabled: true

    hosts: ["localhost:9200"]

    username: "tribe_user"

    password: "tribe_user"

./packetbeat -e -c etc/packetbeat.yml  -d "publish"

2015/12/29 14:24:39.965037 preprocess.go:37: DBG  Start Preprocessing

2015/12/29 14:24:39.965366 publish.go:98: DBG  Publish: {

  "@timestamp": "2015-12-29T14:24:39.709Z",

  "beat": {

    "hostname": "medcls-MacBook.local",

    "name": "medcls-MacBook.local"

  },

  "bytes_in": 31,

  "bytes_out": 115,

  "client_ip": "192.168.3.10",

  "client_port": 53669,

  "client_proc": "",

  "client_server": "",

  "count": 1,

  "direction": "out",

  "dns": {

    "additionals_count": 0,

    "answers": [

      {

        "class": "IN",

        "data": "www.a.shifen.com",

        "name": "sp2.baidu.com",

        "ttl": 333,

        "type": "CNAME"

      }

    ],

    "answers_count": 1,

    "authorities": [

      {

        "class": "IN",

        "data": "ns1.a.shifen.com",

        "expire": 86400,

        "minimum": 3600,

        "name": "a.shifen.com",

        "refresh": 5,

        "retry": 5,

        "rname": "baidu_dns_master.baidu.com",

        "serial": 1512240003,

        "ttl": 12,

        "type": "SOA"

      }

    ],

    "authorities_count": 1,

    "flags": {

      "authoritative": false,

      "recursion_allowed": true,

      "recursion_desired": true,

      "truncated_response": false

    },

    "id": 7435,

    "op_code": "QUERY",

    "question": {

      "class": "IN",

      "name": "sp2.baidu.com",

      "type": "AAAA"

    },

    "response_code": "NOERROR"

  },

  "ip": "192.168.3.1",

  "method": "QUERY",

  "port": 53,

  "proc": "",

  "query": "class IN, type AAAA, sp2.baidu.com",

  "resource": "sp2.baidu.com",

  "responsetime": 18,

  "server": "",

  "status": "OK",

  "transport": "udp",

  "type": "dns"

}

2015/12/29 14:24:39.965774 preprocess.go:94: DBG  Forward preprocessed events

2015/12/29 14:24:39.965796 async.go:42: DBG  async forward to outputers (1)

2015/12/29 14:24:40.099973 output.go:103: DBG  output worker: publish 2 events

curl http://localhost:9200/_cat/indices\?pretty\=true -u tribe_user:tribe_user

yellow open packetbeat-2015.12.29  5 1   135  0 561.2kb 561.2kb

index.query.bool.max_clause_count: 4096

• 如果某个shard主片在，副片所在结点挂了，那么选择另外一个可用结点，将副片分配(allocate)上去，然后进行主从片的复制。
• 如果某个shard的主片所在结点挂了，副片还在，那么将副片升级为主片，然后做主副复制。
• 如果某个shard的主副片所在结点都挂了，则暂时无法恢复，等待持有相关数据的结点重新加入集群后，从结点上恢复主分片，再选择某个结点分配复制片，并从主分片同步数据。

Green: 所有的shard主副片都完好的
Yellow: 所有shard的主片都完好，部分副片没有了，数据完整性依然完好。
Red: 某些shard的主副片都没有了，对应的索引数据不完整

gateway.expected_nodes
gateway.expected_master_nodes
gateway.expected_data_nodes

gateway.recover_after_nodes
gateway.recover_after_master_nodes
gateway.recover_after_data_nodes

gateway.expected_data_nodes: 10
gateway.recover_after_time: 5m
gateway.recover_after_data_nodes: 8

1. 暂停数据写入程序
2. 关闭集群shard allocation
3. 手动执行POST /_flush/synced
4. 重启结点
5. 重新开启集群shard allocation 
6. 等待recovery完成，集群health status变成green
7. 重新开启数据写入程序

1. 对主片上的segment file做一个快照，然后拷贝到复制片分配到的结点。数据拷贝期间，不会阻塞索引请求，新增索引操作记录到translog里。
2. 对translog做一个快照，此快照包含第一阶段新增的索引请求，然后重放快照里的索引操作。此阶段仍然不阻塞索引请求，新增索引操作记录到translog里。
3. 为了能达到主副片完全同步，阻塞掉新索引请求，然后重放阶段二新增的translog操作。

curl -XPUT 'http://127.0.0.1:9200/login-2015.12.23/' -d '{

  "settings" : {

    "number_of_shards" : 1

  },

  "mappings" : {

    "logs" : {

      "properties" : {

        "uid" : { "type" : "string", "index" : "not_analyzed" },

        "register_time" : { "type" : "date", "index" : "not_analyzed" },

        "login_time" : { "type" : "date", "index" : "not_analyzed" }

      }

    }

  }

}'

{"index":{"_index":"login-2015.12.23","_type":"logs"}}

{"uid":"1","register_time":"2015-12-23T12:00:00Z","login_time":"2015-12-23T12:00:00Z"}

{"index":{"_index":"login-2015.12.23","_type":"logs"}}

{"uid":"2","register_time":"2015-12-23T12:00:00Z","login_time":"2015-12-23T12:00:00Z"}

{"index":{"_index":"login-2015.12.24","_type":"logs"}}

{"uid":"1","register_time":"2015-12-23T12:00:00Z","login_time":"2015-12-24T12:00:00Z"}

curl -XPOST 'http://127.0.0.1:9200/login-2015.12.23,login-2015.12.24/_search' -d'

{

  "size" : 0,

  "aggs" : {

    "new_users" : {



      "filters" : {

        "filters" : [

          {

            "range" : {

              "register_time" : {

                "gte" : "2015-12-23",

                "lt" : "2015-12-24"

              }

            }

          }

        ]

      },

      "aggs" : {

        "register_count" : {

          "cardinality" : {

            "field" : "uid"

          }

        },

        "today" : {

          "filter" : {

            "range" : {

              "login_time" : {

                "gte" : "2015-12-24",

                "lt" : "2015-12-25"

              }

            }

          },

          "aggs" : {

            "login_count" : {

              "cardinality" : {

                "field" : "uid"

              }

            }

          }

        },

        "retention" : {

          "bucket_script" : {

            "buckets_path" : {

              "today_count" : "today>login_count",

              "yesterday_count" : "register_count"

            },

            "script" : {

              "lang" : "expression",

              "inline" : "today_count / yesterday_count"

            }

          }

        }

      }

    }

  }

}'

1. pipeline agg 的 parent agg 必须是返回数组的 buckets agg 类型。我这里曾经打算使用 filter agg 直接请求register_time:["now-2d" TO "now-1d"]，结果报错说找不到 buckets_path 的 START_OBJECT。所以改用了 filters agg 的数组格式。
2. bucket_script agg 同样受 scripting module 的影响。也就是说，官网示例里的"script":"today_count / yesterday_count" 这种写法，是采用了 groovy 引擎的 inline 模式。在 ES 2.0 的默认设置下，是被禁止运行的！所以，应该按照 scripting module 的统一要求，改写成 file 形式存放到 config/scripts下；或者改用 Lucene Expression 运行。考虑到 pipeline aggregation 只支持数值运算，这里使用 groovy 价值不大，所以直接指明 lang 参数即可。

{

  "took" : 3,

  "timed_out" : false,

  "_shards" : {

    "total" : 1,

    "successful" : 1,

    "failed" : 0

  },

  "hits" : {

    "total" : 3,

    "max_score" : 0.0,

    "hits" : [ ]

  },

  "aggregations" : {

    "new_users" : {

      "buckets" : [ {

        "doc_count" : 3,

        "today" : {

          "doc_count" : 1,

          "login_count" : {

            "value" : 1

          }

        },

        "register_count" : {

          "value" : 2

        },

        "retention" : {

          "value" : 0.5

        }

      } ]

    }

  }

}

[2015-12-23 23:20:47,749] ERROR [KafkaApi-0] error when handling request Name: TopicMetadataRequest; Version: 0; CorrelationId: 48; ClientId: ; Topics: test-%{type} (kafka.server.KafkaApis)

kafka.common.InvalidTopicException: topic name test-%{type} is illegal, contains a character other than ASCII alphanumerics, '.', '_' and '-'

        at kafka.common.Topic$.validate(Topic.scala:42)

        at kafka.admin.AdminUtils$.createOrUpdateTopicPartitionAssignmentPathInZK(AdminUtils.scala:181)

        at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:172)

        at kafka.server.KafkaApis$$anonfun$19.apply(KafkaApis.scala:520)

        at kafka.server.KafkaApis$$anonfun$19.apply(KafkaApis.scala:503)

        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)

        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)

        at scala.collection.immutable.Set$Set1.foreach(Set.scala:74)

        at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)

        at scala.collection.AbstractSet.scala$collection$SetLike$$super$map(Set.scala:47)

        at scala.collection.SetLike$class.map(SetLike.scala:93)

        at scala.collection.AbstractSet.map(Set.scala:47)

        at kafka.server.KafkaApis.getTopicMetadata(KafkaApis.scala:503)

        at kafka.server.KafkaApis.handleTopicMetadataRequest(KafkaApis.scala:542)

        at kafka.server.KafkaApis.handle(KafkaApis.scala:62)

        at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:59)

        at java.lang.Thread.run(Thread.java:744)

tcpdump -nn 'dst port 9092 and tcp[37]==3 and tcp[57]==37'

tcpdump -nn 'dst port 9092 and tcp[(tcp[12]>>2)+5]==3 and tcp[(tcp[12]>>2)+25]==37'

tcpdump -vv -nn -X -s 0 dst port 9092 | grep -C 5 "test-"

1. 倒排词典的索引需要常驻内存，无法GC，需要监控data node上segment memory增长趋势。
2. 各类缓存，field cache, filter cache, indexing cache, bulk queue等等，要设置合理的大小，并且要应该根据最坏的情况来看heap是否够用，也就是各类缓存全部占满的时候，还有heap空间可以分配给其他任务吗？避免采用clear cache等“自欺欺人”的方式来释放内存。
3. 避免返回大量结果集的搜索与聚合。确实需要大量拉取数据的场景，可以采用scan & scroll api来实现。
4. cluster stats驻留内存并无法水平扩展，超大规模集群可以考虑分拆成多个集群通过tribe node连接。
5. 想知道heap够不够，必须结合实际应用场景，并对集群的heap使用情况做持续的监控。
6. 根据监控数据理解内存需求，合理配置各类circuit breaker，将内存溢出风险降低到最低。