本文整理汇总了Golang中github.com/bitly/nsq/util.ApiRequest函数的典型用法代码示例。如果您正苦于以下问题:Golang ApiRequest函数的具体用法?Golang ApiRequest怎么用?Golang ApiRequest使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了ApiRequest函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Golang代码示例。
示例1: GetLookupdTopicChannels
// GetLookupdTopicChannels returns a []string containing a union of the channels
// from all the given lookupd for the given topic
func GetLookupdTopicChannels(topic string, lookupdHTTPAddrs []string) ([]string, error) {
success := false
allChannels := make([]string, 0)
var lock sync.Mutex
var wg sync.WaitGroup
for _, addr := range lookupdHTTPAddrs {
wg.Add(1)
endpoint := fmt.Sprintf("http://%s/channels?topic=%s", addr, url.QueryEscape(topic))
log.Printf("LOOKUPD: querying %s", endpoint)
go func(endpoint string) {
data, err := util.ApiRequest(endpoint)
lock.Lock()
defer lock.Unlock()
defer wg.Done()
if err != nil {
log.Printf("ERROR: lookupd %s - %s", endpoint, err.Error())
return
}
success = true
// {"data":{"channels":["test"]}}
channels, _ := data.Get("channels").StringArray()
allChannels = util.StringUnion(allChannels, channels)
}(endpoint)
}
wg.Wait()
sort.Strings(allChannels)
if success == false {
return nil, errors.New("unable to query any lookupd")
}
return allChannels, nil
}
开发者ID:johntdyer,项目名称:golang-devops-stuff,代码行数:33,代码来源:lookupd.go
示例2: deleteTopicHandler
func (s *httpServer) deleteTopicHandler(w http.ResponseWriter, req *http.Request) {
if req.Method != "POST" {
log.Printf("ERROR: invalid %s to POST only method", req.Method)
http.Error(w, "INVALID_REQUEST", 500)
return
}
reqParams := &util.PostParams{req}
topicName, err := reqParams.Get("topic")
if err != nil {
http.Error(w, "MISSING_ARG_TOPIC", 500)
return
}
rd, _ := reqParams.Get("rd")
if !strings.HasPrefix(rd, "/") {
rd = "/"
}
// for topic removal, you need to get all the producers *first*
producers := s.getProducers(topicName)
// remove the topic from all the lookupds
for _, addr := range s.context.nsqadmin.options.NSQLookupdHTTPAddresses {
endpoint := fmt.Sprintf("http://%s/delete_topic?topic=%s", addr, url.QueryEscape(topicName))
log.Printf("LOOKUPD: querying %s", endpoint)
_, err := util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: lookupd %s - %s", endpoint, err.Error())
continue
}
}
// now remove the topic from all the producers
for _, addr := range producers {
endpoint := fmt.Sprintf("http://%s/delete_topic?topic=%s", addr, url.QueryEscape(topicName))
log.Printf("NSQD: querying %s", endpoint)
_, err := util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: nsqd %s - %s", endpoint, err.Error())
continue
}
}
s.notifyAdminAction("delete_topic", topicName, "", "", req)
http.Redirect(w, req, rd, 302)
}
开发者ID:jsocol,项目名称:nsq,代码行数:49,代码来源:http.go
示例3: tombstoneTopicProducerHandler
func (s *httpServer) tombstoneTopicProducerHandler(w http.ResponseWriter, req *http.Request) {
if req.Method != "POST" {
log.Printf("ERROR: invalid %s to POST only method", req.Method)
http.Error(w, "INVALID_REQUEST", 500)
return
}
reqParams := &util.PostParams{req}
topicName, err := reqParams.Get("topic")
if err != nil {
http.Error(w, "MISSING_ARG_TOPIC", 500)
return
}
node, err := reqParams.Get("node")
if err != nil {
http.Error(w, "MISSING_ARG_NODE", 500)
return
}
rd, _ := reqParams.Get("rd")
if !strings.HasPrefix(rd, "/") {
rd = "/"
}
// tombstone the topic on all the lookupds
for _, addr := range s.context.nsqadmin.options.NSQLookupdHTTPAddresses {
endpoint := fmt.Sprintf("http://%s/tombstone_topic_producer?topic=%s&node=%s",
addr, url.QueryEscape(topicName), url.QueryEscape(node))
log.Printf("LOOKUPD: querying %s", endpoint)
_, err := util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: lookupd %s - %s", endpoint, err.Error())
}
}
// delete the topic on the producer
endpoint := fmt.Sprintf("http://%s/delete_topic?topic=%s", node, url.QueryEscape(topicName))
log.Printf("NSQD: querying %s", endpoint)
_, err = util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: nsqd %s - %s", endpoint, err.Error())
}
s.notifyAdminAction("tombstone_topic_producer", topicName, "", node, req)
http.Redirect(w, req, rd, 302)
}
开发者ID:jsocol,项目名称:nsq,代码行数:48,代码来源:http.go
示例4: GetNSQDTopics
// GetNSQDTopics returns a []string containing all the topics
// produced by the given nsqd
func GetNSQDTopics(nsqdHTTPAddrs []string) ([]string, error) {
topics := make([]string, 0)
var lock sync.Mutex
var wg sync.WaitGroup
success := false
for _, addr := range nsqdHTTPAddrs {
wg.Add(1)
endpoint := fmt.Sprintf("http://%s/stats?format=json", addr)
log.Printf("NSQD: querying %s", endpoint)
go func(endpoint string) {
data, err := util.ApiRequest(endpoint)
lock.Lock()
defer lock.Unlock()
defer wg.Done()
if err != nil {
log.Printf("ERROR: lookupd %s - %s", endpoint, err.Error())
return
}
success = true
topicList, _ := data.Get("topics").Array()
for i := range topicList {
topicInfo := data.Get("topics").GetIndex(i)
topics = util.StringAdd(topics, topicInfo.Get("topic_name").MustString())
}
}(endpoint)
}
wg.Wait()
sort.Strings(topics)
if success == false {
return nil, errors.New("unable to query any nsqd")
}
return topics, nil
}
开发者ID:johntdyer,项目名称:golang-devops-stuff,代码行数:36,代码来源:lookupd.go
示例5: emptyTopicHandler
func (s *httpServer) emptyTopicHandler(w http.ResponseWriter, req *http.Request) {
if req.Method != "POST" {
log.Printf("ERROR: invalid %s to POST only method", req.Method)
http.Error(w, "INVALID_REQUEST", 500)
return
}
reqParams := &util.PostParams{req}
topicName, err := reqParams.Get("topic")
if err != nil {
http.Error(w, "MISSING_ARG_TOPIC", 500)
return
}
producers := s.getProducers(topicName)
for _, addr := range producers {
endpoint := fmt.Sprintf("http://%s/empty_topic?topic=%s", addr, url.QueryEscape(topicName))
log.Printf("NSQD: calling %s", endpoint)
_, err := util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: nsqd %s - %s", endpoint, err.Error())
continue
}
}
s.notifyAdminAction("empty_topic", topicName, "", "", req)
http.Redirect(w, req, fmt.Sprintf("/topic/%s", url.QueryEscape(topicName)), 302)
}
开发者ID:jsocol,项目名称:nsq,代码行数:30,代码来源:http.go
示例6: TestTombstonedNodes
func TestTombstonedNodes(t *testing.T) {
log.SetOutput(ioutil.Discard)
defer log.SetOutput(os.Stdout)
tcpAddr, httpAddr, nsqlookupd := mustStartLookupd()
defer nsqlookupd.Exit()
nsqlookupd.inactiveProducerTimeout = 50 * time.Millisecond
lookupdHTTPAddrs := []string{fmt.Sprintf("%s", httpAddr)}
topicName := "inactive_nodes"
conn := mustConnectLookupd(t, tcpAddr)
identify(t, conn, "ip.address", 5000, 5555, "fake-version")
nsq.Register(topicName, "channel1").Write(conn)
_, err := nsq.ReadResponse(conn)
assert.Equal(t, err, nil)
producers, _ := lookuputil.GetLookupdProducers(lookupdHTTPAddrs)
assert.Equal(t, len(producers), 1)
assert.Equal(t, len(producers[0].Topics), 1)
assert.Equal(t, producers[0].Topics[0].Topic, topicName)
assert.Equal(t, producers[0].Topics[0].Tombstoned, false)
endpoint := fmt.Sprintf("http://%s/tombstone_topic_producer?topic=%s&node=%s", httpAddr, topicName, "ip.address:5555")
_, err = util.ApiRequest(endpoint)
assert.Equal(t, err, nil)
producers, _ = lookuputil.GetLookupdProducers(lookupdHTTPAddrs)
assert.Equal(t, len(producers), 1)
assert.Equal(t, len(producers[0].Topics), 1)
assert.Equal(t, producers[0].Topics[0].Topic, topicName)
assert.Equal(t, producers[0].Topics[0].Tombstoned, true)
}
开发者ID:michaelhood,项目名称:nsq,代码行数:35,代码来源:nsqlookupd_test.go
示例7: pauseChannelHandler
func (s *httpServer) pauseChannelHandler(w http.ResponseWriter, req *http.Request) {
if req.Method != "POST" {
log.Printf("ERROR: invalid %s to POST only method", req.Method)
http.Error(w, "INVALID_REQUEST", 500)
return
}
reqParams := &util.PostParams{req}
topicName, channelName, err := util.GetTopicChannelArgs(reqParams)
if err != nil {
http.Error(w, err.Error(), 500)
return
}
producers := s.getProducers(topicName)
for _, addr := range producers {
endpoint := fmt.Sprintf("http://%s%s?topic=%s&channel=%s",
addr, req.URL.Path, url.QueryEscape(topicName), url.QueryEscape(channelName))
log.Printf("NSQD: calling %s", endpoint)
_, err := util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: nsqd %s - %s", endpoint, err.Error())
continue
}
}
s.notifyAdminAction(strings.TrimLeft(req.URL.Path, "/"), topicName, channelName, "", req)
http.Redirect(w, req, fmt.Sprintf("/topic/%s/%s", url.QueryEscape(topicName), url.QueryEscape(channelName)), 302)
}
开发者ID:jsocol,项目名称:nsq,代码行数:31,代码来源:http.go
示例8: CreateTopicChannel
// CreateTopicChannel creates the given topic and channel on the provided
// nsqd instance
func CreateTopicChannel(topic, channel, nsqdAddr string) (err error) {
endpoint := fmt.Sprintf("http://%s/create_topic?topic=%s", nsqdAddr, url.QueryEscape(topic))
_, err = util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: Could not create topic %s: %s", topic, err)
return
}
endpoint = fmt.Sprintf("http://%s/create_channel?topic=%s&channel=%s", nsqdAddr, url.QueryEscape(topic), url.QueryEscape(channel))
_, err = util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: Could not create channel %s: %s", channel, err)
return
}
return
}
开发者ID:kzvezdarov,项目名称:alexander,代码行数:18,代码来源:message_node.go
示例9: deleteChannelHandler
func (s *httpServer) deleteChannelHandler(w http.ResponseWriter, req *http.Request) {
if req.Method != "POST" {
log.Printf("ERROR: invalid %s to POST only method", req.Method)
http.Error(w, "INVALID_REQUEST", 500)
return
}
reqParams := &util.PostParams{req}
topicName, channelName, err := util.GetTopicChannelArgs(reqParams)
if err != nil {
http.Error(w, err.Error(), 500)
return
}
rd, _ := reqParams.Get("rd")
if !strings.HasPrefix(rd, "/") {
rd = fmt.Sprintf("/topic/%s", url.QueryEscape(topicName))
}
for _, addr := range s.context.nsqadmin.options.NSQLookupdHTTPAddresses {
endpoint := fmt.Sprintf("http://%s/delete_channel?topic=%s&channel=%s",
addr, url.QueryEscape(topicName), url.QueryEscape(channelName))
log.Printf("LOOKUPD: querying %s", endpoint)
_, err := util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: lookupd %s - %s", endpoint, err.Error())
continue
}
}
producers := s.getProducers(topicName)
for _, addr := range producers {
endpoint := fmt.Sprintf("http://%s/delete_channel?topic=%s&channel=%s",
addr, url.QueryEscape(topicName), url.QueryEscape(channelName))
log.Printf("NSQD: querying %s", endpoint)
_, err := util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: nsqd %s - %s", endpoint, err.Error())
continue
}
}
s.notifyAdminAction("delete_channel", topicName, channelName, "", req)
http.Redirect(w, req, rd, 302)
}
开发者ID:jsocol,项目名称:nsq,代码行数:47,代码来源:http.go
示例10: TestTombstoneRecover
func TestTombstoneRecover(t *testing.T) {
log.SetOutput(ioutil.Discard)
defer log.SetOutput(os.Stdout)
options := NewNSQLookupdOptions()
options.TombstoneLifetime = 50 * time.Millisecond
tcpAddr, httpAddr, nsqlookupd := mustStartLookupd(options)
defer nsqlookupd.Exit()
topicName := "tombstone_recover"
topicName2 := topicName + "2"
conn := mustConnectLookupd(t, tcpAddr)
identify(t, conn, "ip.address", 5000, 5555, "fake-version")
nsq.Register(topicName, "channel1").WriteTo(conn)
_, err := nsq.ReadResponse(conn)
assert.Equal(t, err, nil)
nsq.Register(topicName2, "channel2").WriteTo(conn)
_, err = nsq.ReadResponse(conn)
assert.Equal(t, err, nil)
endpoint := fmt.Sprintf("http://%s/tombstone_topic_producer?topic=%s&node=%s", httpAddr, topicName, "ip.address:5555")
_, err = util.ApiRequest(endpoint)
assert.Equal(t, err, nil)
endpoint = fmt.Sprintf("http://%s/lookup?topic=%s", httpAddr, topicName)
data, err := util.ApiRequest(endpoint)
assert.Equal(t, err, nil)
producers, _ := data.Get("producers").Array()
assert.Equal(t, len(producers), 0)
endpoint = fmt.Sprintf("http://%s/lookup?topic=%s", httpAddr, topicName2)
data, err = util.ApiRequest(endpoint)
assert.Equal(t, err, nil)
producers, _ = data.Get("producers").Array()
assert.Equal(t, len(producers), 1)
time.Sleep(55 * time.Millisecond)
endpoint = fmt.Sprintf("http://%s/lookup?topic=%s", httpAddr, topicName)
data, err = util.ApiRequest(endpoint)
assert.Equal(t, err, nil)
producers, _ = data.Get("producers").Array()
assert.Equal(t, len(producers), 1)
}
开发者ID:pombredanne,项目名称:nsq,代码行数:47,代码来源:nsqlookupd_test.go
示例11: DeleteTopic
// DeleteTopic deletes the given topic from the provided nsqlookupd instance
func DeleteTopic(topic, nsqlookupdHttpAddr string) (err error) {
endpoint := fmt.Sprintf("http://%s/delete_topic?topic=%s", nsqlookupdHttpAddr, url.QueryEscape(topic))
_, err = util.ApiRequest(endpoint)
if err != nil {
log.Printf("ERROR: Could not delete topic %s: %s", topic, err)
}
return
}
开发者ID:kzvezdarov,项目名称:alexander,代码行数:9,代码来源:message_node.go
示例12: TestChannelUnregister
func TestChannelUnregister(t *testing.T) {
log.SetOutput(ioutil.Discard)
defer log.SetOutput(os.Stdout)
tcpAddr, httpAddr, nsqlookupd := mustStartLookupd(NewNSQLookupdOptions())
defer nsqlookupd.Exit()
topics := nsqlookupd.DB.FindRegistrations("topic", "*", "*")
assert.Equal(t, len(topics), 0)
topicName := "channel_unregister"
conn := mustConnectLookupd(t, tcpAddr)
tcpPort := 5000
httpPort := 5555
identify(t, conn, "ip.address", tcpPort, httpPort, "fake-version")
nsq.Register(topicName, "ch1").WriteTo(conn)
v, err := nsq.ReadResponse(conn)
assert.Equal(t, err, nil)
assert.Equal(t, v, []byte("OK"))
topics = nsqlookupd.DB.FindRegistrations("topic", topicName, "")
assert.Equal(t, len(topics), 1)
channels := nsqlookupd.DB.FindRegistrations("channel", topicName, "*")
assert.Equal(t, len(channels), 1)
nsq.UnRegister(topicName, "ch1").WriteTo(conn)
v, err = nsq.ReadResponse(conn)
assert.Equal(t, err, nil)
assert.Equal(t, v, []byte("OK"))
topics = nsqlookupd.DB.FindRegistrations("topic", topicName, "")
assert.Equal(t, len(topics), 1)
// we should still have mention of the topic even though there is no producer
// (ie. we haven't *deleted* the channel, just unregistered as a producer)
channels = nsqlookupd.DB.FindRegistrations("channel", topicName, "*")
assert.Equal(t, len(channels), 1)
endpoint := fmt.Sprintf("http://%s/lookup?topic=%s", httpAddr, topicName)
data, err := util.ApiRequest(endpoint)
assert.Equal(t, err, nil)
returnedProducers, err := data.Get("producers").Array()
assert.Equal(t, err, nil)
assert.Equal(t, len(returnedProducers), 1)
}
开发者ID:pombredanne,项目名称:nsq,代码行数:48,代码来源:nsqlookupd_test.go
示例13: TestClientAttributes
func TestClientAttributes(t *testing.T) {
log.SetOutput(ioutil.Discard)
defer log.SetOutput(os.Stdout)
userAgent := "Test User Agent"
options := NewNSQDOptions()
options.Verbose = true
options.SnappyEnabled = true
tcpAddr, httpAddr, nsqd := mustStartNSQD(options)
defer nsqd.Exit()
conn, err := mustConnectNSQD(tcpAddr)
assert.Equal(t, err, nil)
data := identify(t, conn, map[string]interface{}{
"snappy": true,
"user_agent": userAgent,
}, nsq.FrameTypeResponse)
resp := struct {
Snappy bool `json:"snappy"`
UserAgent string `json:"user_agent"`
}{}
err = json.Unmarshal(data, &resp)
assert.Equal(t, err, nil)
assert.Equal(t, resp.Snappy, true)
r := snappystream.NewReader(conn, snappystream.SkipVerifyChecksum)
w := snappystream.NewWriter(conn)
topicName := "test_client_attributes" + strconv.Itoa(int(time.Now().Unix()))
sub(t, readWriter{r, w}, topicName, "ch")
// need to give nsqd a chance to sub the client
time.Sleep(50 * time.Millisecond)
testUrl := fmt.Sprintf("http://127.0.0.1:%d/stats?format=json", httpAddr.Port)
statsData, err := util.ApiRequest(testUrl)
if err != nil {
t.Fatalf(err.Error())
}
client := statsData.Get("topics").GetIndex(0).Get("channels").GetIndex(0).Get("clients").GetIndex(0)
assert.Equal(t, client.Get("user_agent").MustString(), userAgent)
assert.Equal(t, client.Get("snappy").MustBool(), true)
}
开发者ID:hpf311,项目名称:nsq,代码行数:46,代码来源:stats_test.go
示例14: TestClientAttributes
func TestClientAttributes(t *testing.T) {
log.SetOutput(ioutil.Discard)
defer log.SetOutput(os.Stdout)
userAgent := "Test User Agent"
*verbose = true
options := NewNSQDOptions()
options.SnappyEnabled = true
tcpAddr, httpAddr, nsqd := mustStartNSQD(options)
defer nsqd.Exit()
conn, err := mustConnectNSQD(tcpAddr)
assert.Equal(t, err, nil)
data := identifyFeatureNegotiation(t, conn, map[string]interface{}{"snappy": true, "user_agent": userAgent})
r := struct {
Snappy bool `json:"snappy"`
UserAgent string `json:"user_agent"`
}{}
err = json.Unmarshal(data, &r)
assert.Equal(t, err, nil)
assert.Equal(t, r.Snappy, true)
compressConn := snappystream.NewReader(conn, snappystream.SkipVerifyChecksum)
w := snappystream.NewWriter(conn)
rw := readWriter{compressConn, w}
topicName := "test_client_attributes" + strconv.Itoa(int(time.Now().Unix()))
sub(t, rw, topicName, "ch")
err = nsq.Ready(1).Write(rw)
assert.Equal(t, err, nil)
testUrl := fmt.Sprintf("http://127.0.0.1:%d/stats?format=json", httpAddr.Port)
statsData, err := util.ApiRequest(testUrl)
if err != nil {
t.Fatalf(err.Error())
}
client := statsData.Get("topics").GetIndex(0).Get("channels").GetIndex(0).Get("clients").GetIndex(0)
assert.Equal(t, client.Get("user_agent").MustString(), userAgent)
assert.Equal(t, client.Get("snappy").MustBool(), true)
}
开发者ID:johntdyer,项目名称:golang-devops-stuff,代码行数:46,代码来源:stats_test.go
示例15: GetLookupdTopicProducers
// GetLookupdTopicProducers returns a []string of the broadcast_address:http_port of all the
// producers for a given topic by unioning the results returned from the given lookupd
func GetLookupdTopicProducers(topic string, lookupdHTTPAddrs []string) ([]string, error) {
success := false
allSources := make([]string, 0)
var lock sync.Mutex
var wg sync.WaitGroup
for _, addr := range lookupdHTTPAddrs {
wg.Add(1)
endpoint := fmt.Sprintf("http://%s/lookup?topic=%s", addr, url.QueryEscape(topic))
log.Printf("LOOKUPD: querying %s", endpoint)
go func(endpoint string) {
data, err := util.ApiRequest(endpoint)
lock.Lock()
defer lock.Unlock()
defer wg.Done()
if err != nil {
log.Printf("ERROR: lookupd %s - %s", endpoint, err.Error())
return
}
success = true
producers := data.Get("producers")
producersArray, _ := producers.Array()
for i := range producersArray {
producer := producers.GetIndex(i)
address := producer.Get("address").MustString() //TODO: remove for 1.0
broadcastAddress := producer.Get("broadcast_address").MustString()
if broadcastAddress == "" {
broadcastAddress = address
}
httpPort := producer.Get("http_port").MustInt()
key := fmt.Sprintf("%s:%d", broadcastAddress, httpPort)
allSources = util.StringAdd(allSources, key)
}
}(endpoint)
}
wg.Wait()
if success == false {
return nil, errors.New("unable to query any lookupd")
}
return allSources, nil
}
开发者ID:michaelhood,项目名称:nsq,代码行数:45,代码来源:lookupd.go
示例16: GetNSQDTopicProducers
// GetNSQDTopicProducers returns a []string containing the addresses of all the nsqd
// that produce the given topic out of the given nsqd
func GetNSQDTopicProducers(topic string, nsqdHTTPAddrs []string) ([]string, error) {
addresses := make([]string, 0)
var lock sync.Mutex
var wg sync.WaitGroup
success := false
for _, addr := range nsqdHTTPAddrs {
wg.Add(1)
endpoint := fmt.Sprintf("http://%s/stats?format=json", addr)
log.Printf("NSQD: querying %s", endpoint)
go func(endpoint string) {
data, err := util.ApiRequest(endpoint)
lock.Lock()
defer lock.Unlock()
defer wg.Done()
if err != nil {
log.Printf("ERROR: lookupd %s - %s", endpoint, err.Error())
return
}
success = true
topicList, _ := data.Get("topics").Array()
for _, topicInfo := range topicList {
topicInfo := topicInfo.(map[string]interface{})
topicName := topicInfo["topic_name"].(string)
if topicName == topic {
addresses = append(addresses, addr)
return
}
}
}(endpoint)
}
wg.Wait()
if success == false {
return nil, errors.New("unable to query any nsqd")
}
return addresses, nil
}
开发者ID:michaelhood,项目名称:nsq,代码行数:39,代码来源:lookupd.go
示例17: GetNSQDStats
// GetNSQDStats returns aggregate topic and channel stats from the given NSQD instances
//
// if selectedTopic is empty, this will return stats for *all* topic/channels
// and the ChannelStats dict will be keyed by topic + ':' + channel
func GetNSQDStats(nsqdHTTPAddrs []string, selectedTopic string) ([]*TopicStats, map[string]*ChannelStats, error) {
var lock sync.Mutex
var wg sync.WaitGroup
topicStatsList := make(TopicStatsList, 0)
channelStatsMap := make(map[string]*ChannelStats)
success := false
for _, addr := range nsqdHTTPAddrs {
wg.Add(1)
endpoint := fmt.Sprintf("http://%s/stats?format=json", addr)
log.Printf("NSQD: querying %s", endpoint)
go func(endpoint string, addr string) {
data, err := util.ApiRequest(endpoint)
lock.Lock()
defer lock.Unlock()
defer wg.Done()
if err != nil {
log.Printf("ERROR: lookupd %s - %s", endpoint, err.Error())
return
}
success = true
topics, _ := data.Get("topics").Array()
for _, t := range topics {
t := t.(map[string]interface{})
topicName := t["topic_name"].(string)
if selectedTopic != "" && topicName != selectedTopic {
continue
}
depth := int64(t["depth"].(float64))
backendDepth := int64(t["backend_depth"].(float64))
topicStats := &TopicStats{
HostAddress: addr,
TopicName: topicName,
Depth: depth,
BackendDepth: backendDepth,
MemoryDepth: depth - backendDepth,
MessageCount: int64(t["message_count"].(float64)),
ChannelCount: len(t["channels"].([]interface{})),
}
topicStatsList = append(topicStatsList, topicStats)
channels := t["channels"].([]interface{})
for _, c := range channels {
c := c.(map[string]interface{})
channelName := c["channel_name"].(string)
key := channelName
if selectedTopic == "" {
key = fmt.Sprintf("%s:%s", topicName, channelName)
}
channelStats, ok := channelStatsMap[key]
if !ok {
channelStats = &ChannelStats{
HostAddress: addr,
TopicName: topicName,
ChannelName: channelName,
}
channelStatsMap[key] = channelStats
}
depth := int64(c["depth"].(float64))
backendDepth := int64(c["backend_depth"].(float64))
var paused bool
pausedInterface, ok := c["paused"]
if ok {
paused = pausedInterface.(bool)
}
clients := c["clients"].([]interface{})
hostChannelStats := &ChannelStats{
HostAddress: addr,
TopicName: topicName,
ChannelName: channelName,
Depth: depth,
BackendDepth: backendDepth,
MemoryDepth: depth - backendDepth,
Paused: paused,
InFlightCount: int64(c["in_flight_count"].(float64)),
DeferredCount: int64(c["deferred_count"].(float64)),
MessageCount: int64(c["message_count"].(float64)),
RequeueCount: int64(c["requeue_count"].(float64)),
TimeoutCount: int64(c["timeout_count"].(float64)),
// TODO: this is sort of wrong; clients should be de-duped
// client A that connects to NSQD-a and NSQD-b should only be counted once. right?
ClientCount: len(clients),
}
channelStats.Add(hostChannelStats)
for _, client := range clients {
//.........这里部分代码省略.........
开发者ID:michaelhood,项目名称:nsq,代码行数:101,代码来源:lookupd.go
示例18: TestBasicLookupd
func TestBasicLookupd(t *testing.T) {
log.SetOutput(ioutil.Discard)
defer log.SetOutput(os.Stdout)
tcpAddr, httpAddr, nsqlookupd := mustStartLookupd(NewNSQLookupdOptions())
defer nsqlookupd.Exit()
topics := nsqlookupd.DB.FindRegistrations("topic", "*", "*")
assert.Equal(t, len(topics), 0)
topicName := "connectmsg"
conn := mustConnectLookupd(t, tcpAddr)
tcpPort := 5000
httpPort := 5555
identify(t, conn, "ip.address", tcpPort, httpPort, "fake-version")
nsq.Register(topicName, "channel1").WriteTo(conn)
v, err := nsq.ReadResponse(conn)
assert.Equal(t, err, nil)
assert.Equal(t, v, []byte("OK"))
endpoint := fmt.Sprintf("http://%s/nodes", httpAddr)
data, err := util.ApiRequest(endpoint)
log.Printf("got %v", data)
returnedProducers, err := data.Get("producers").Array()
assert.Equal(t, err, nil)
assert.Equal(t, len(returnedProducers), 1)
topics = nsqlookupd.DB.FindRegistrations("topic", topicName, "")
assert.Equal(t, len(topics), 1)
producers := nsqlookupd.DB.FindProducers("topic", topicName, "")
assert.Equal(t, len(producers), 1)
producer := producers[0]
assert.Equal(t, producer.peerInfo.BroadcastAddress, "ip.address")
assert.Equal(t, producer.peerInfo.Hostname, "ip.address")
assert.Equal(t, producer.peerInfo.TcpPort, tcpPort)
assert.Equal(t, producer.peerInfo.HttpPort, httpPort)
endpoint = fmt.Sprintf("http://%s/topics", httpAddr)
data, err = util.ApiRequest(endpoint)
assert.Equal(t, err, nil)
returnedTopics, err := data.Get("topics").Array()
log.Printf("got returnedTopics %v", returnedTopics)
assert.Equal(t, err, nil)
assert.Equal(t, len(returnedTopics), 1)
endpoint = fmt.Sprintf("http://%s/lookup?topic=%s", httpAddr, topicName)
data, err = util.ApiRequest(endpoint)
assert.Equal(t, err, nil)
returnedChannels, err := data.Get("channels").Array()
assert.Equal(t, err, nil)
assert.Equal(t, len(returnedChannels), 1)
returnedProducers, err = data.Get("producers").Array()
log.Printf("got returnedProducers %v", returnedProducers)
assert.Equal(t, err, nil)
assert.Equal(t, len(returnedProducers), 1)
for i := range returnedProducers {
producer := data.Get("producers").GetIndex(i)
log.Printf("producer %v", producer)
port, err := producer.Get("tcp_port").Int()
assert.Equal(t, err, nil)
assert.Equal(t, port, tcpPort)
port, err = producer.Get("http_port").Int()
assert.Equal(t, err, nil)
assert.Equal(t, port, httpPort)
broadcastaddress, err := producer.Get("broadcast_address").String()
assert.Equal(t, err, nil)
assert.Equal(t, broadcastaddress, "ip.address")
ver, err := producer.Get("version").String()
assert.Equal(t, err, nil)
assert.Equal(t, ver, "fake-version")
}
conn.Close()
time.Sleep(10 * time.Millisecond)
// now there should be no producers, but still topic/channel entries
data, err = util.ApiRequest(endpoint)
assert.Equal(t, err, nil)
returnedChannels, err = data.Get("channels").Array()
assert.Equal(t, err, nil)
assert.Equal(t, len(returnedChannels), 1)
returnedProducers, err = data.Get("producers").Array()
assert.Equal(t, err, nil)
assert.Equal(t, len(returnedProducers), 0)
}
开发者ID:pombredanne,项目名称:nsq,代码行数:94,代码来源:nsqlookupd_test.go
示例19: TestNsqdToLookupd
func TestNsqdToLookupd(t *testing.T) {
log.SetOutput(ioutil.Discard)
defer log.SetOutput(os.Stdout)
topicName := "cluster_test" + strconv.Itoa(int(time.Now().Unix()))
hostname, err := os.Hostname()
if err != nil {
t.Fatalf("ERROR: failed to get hostname - %s", err.Error())
}
_, err = util.ApiRequest(fmt.Sprintf("http://127.0.0.1:4151/create_topic?topic=%s", topicName))
if err != nil {
t.Fatalf(err.Error())
}
_, err = util.ApiRequest(fmt.Sprintf("http://127.0.0.1:4151/create_channel?topic=%s&channel=ch", topicName))
if err != nil {
t.Fatalf(err.Error())
}
// allow some time for nsqd to push info to nsqlookupd
time.Sleep(350 * time.Millisecond)
data, err := util.ApiRequest("http://127.0.0.1:4161/debug")
if err != nil {
t.Fatalf(err.Error())
}
producers, _ := data.Get("topic:" + topicName + ":").Array()
assert.Equal(t, len(producers), 1)
producer := producers[0]
producerData, _ := producer.(map[string]interface{})
address := producerData["address"].(string) //TODO: remove for 1.0
producerHostname := producerData["hostname"].(string)
broadcastAddress := producerData["broadcast_address"].(string)
port := int(producerData["tcp_port"].(float64))
tombstoned := producerData["tombstoned"].(bool)
assert.Equal(t, address, hostname)
assert.Equal(t, producerHostname, hostname)
assert.Equal(t, broadcastAddress, hostname)
assert.Equal(t, port, 4150)
assert.Equal(t, tombstoned, false)
producers, _ = data.Get("channel:" + topicName + ":ch").Array()
assert.Equal(t, len(producers), 1)
producer = producers[0]
producerData, _ = producer.(map[string]interface{})
address = producerData["address"].(string) //TODO: remove for 1.0
producerHostname = producerData["hostname"].(string)
broadcastAddress = producerData["broadcast_address"].(string)
port = int(producerData["tcp_port"].(float64))
tombstoned = producerData["tombstoned"].(bool)
assert.Equal(t, address, hostname)
assert.Equal(t, producerHostname, hostname)
assert.Equal(t, broadcastAddress, hostname)
assert.Equal(t, port, 4150)
assert.Equal(t, tombstoned, false)
data, err = util.ApiRequest("http://127.0.0.1:4161/lookup?topic=" + topicName)
if err != nil {
t.Fatalf(err.Error())
}
producers, _ = data.Get("producers").Array()
assert.Equal(t, len(producers), 1)
producer = producers[0]
producerData, _ = producer.(map[string]interface{})
address = producerData["address"].(string) //TODO: remove for 1.0
producerHostname = producerData["hostname"].(string)
broadcastAddress = producerData["broadcast_address"].(string)
port = int(producerData["tcp_port"].(float64))
assert.Equal(t, address, hostname)
assert.Equal(t, producerHostname, hostname)
assert.Equal(t, broadcastAddress, hostname)
assert.Equal(t, port, 4150)
channels, _ := data.Get("channels").Array()
assert.Equal(t, len(channels), 1)
channel := channels[0].(string)
assert.Equal(t, channel, "ch")
data, err = util.ApiRequest("http://127.0.0.1:4151/delete_topic?topic=" + topicName)
if err != nil {
t.Fatalf(err.Error())
}
// allow some time for nsqd to push info to nsqlookupd
time.Sleep(350 * time.Millisecond)
data, err = util.ApiRequest("http://127.0.0.1:4161/lookup?topic=" + topicName)
if err != nil {
t.Fatalf(err.Error())
}
//.........这里部分代码省略.........
开发者ID:michaelhood,项目名称:nsq,代码行数:101,代码来源:cluster_test.go
示例20: GetNSQDStats
// GetNSQDStats returns aggregate topic and channel stats from the given NSQD instances
//
// if selectedTopic is empty, this will return stats for *all* topic/channels
// and the ChannelStats dict will be keyed by topic + ':' + channel
func GetNSQDStats(nsqdHTTPAddrs []string, selectedTopic string) ([]*TopicStats, map[string]*ChannelStats, error) {
var lock sync.Mutex
var wg sync.WaitGroup
topicStatsList := make(TopicStatsList, 0)
channelStatsMap := make(map[string]*ChannelStats)
success := false
for _, addr := range nsqdHTTPAddrs {
wg.Add(1)
endpoint := fmt.Sprintf("http://%s/stats?format=json", addr)
log.Printf("NSQD: querying %s", endpoint)
go func(endpoint string, addr string) {
data, err := util.ApiRequest(endpoint)
lock.Lock()
defer lock.Unlock()
defer wg.Done()
if err != nil {
log.Printf("ERROR: lookupd %s - %s", endpoint, err.Error())
return
}
success = true
topics, _ := data.Get("topics").Array()
for i := range topics {
t := data.Get("topics").GetIndex(i)
topicName := t.Get("topic_name").MustString()
if selectedTopic != "" && topicName != selectedTopic {
continue
}
depth := t.Get("depth").MustInt64()
backendDepth := t.Get("backend_depth").MustInt64()
channels := t.Get("channels").MustArray()
e2eProcessingLatency := util.E2eProcessingLatencyAggregateFromJson(t.Get("e2e_processing_latency"), topicName, "", addr)
topicStats := &TopicStats{
HostAddress: addr,
TopicName: topicName,
Depth: depth,
BackendDepth: backendDepth,
MemoryDepth: depth - backendDepth,
MessageCount: t.Get("message_count").MustInt64(),
ChannelCount: len(channels),
Paused: t.Get("paused").MustBool(),
E2eProcessingLatency: e2eProcessingLatency,
}
topicStatsList = append(topicStatsList, topicStats)
for j := range channels {
c := t.Get("channels").GetIndex(j)
channelName := c.Get("channel_name").MustString()
key := channelName
if selectedTopic == "" {
key = fmt.Sprintf("%s:%s", topicName, channelName)
}
channelStats, ok := channelStatsMap[key]
if !ok {
channelStats = &ChannelStats{
HostAddress: addr,
TopicName: topicName,
ChannelName: channelName,
}
channelStatsMap[key] = channelStats
}
depth := c.Get("depth").MustInt64()
backendDepth := c.Get("backend_depth").MustInt64()
clients := c.Get("clients").MustArray()
e2eProcessingLatency := util.E2eProcessingLatencyAggregateFromJson(c.Get("e2e_processing_latency"), topicName, channelName, addr)
hostChannelStats := &ChannelStats{
HostAddress: addr,
TopicName: topicName,
ChannelName: channelName,
Depth: depth,
BackendDepth: backendDepth,
MemoryDepth: depth - backendDepth,
Paused: c.Get("paused").MustBool(),
InFlightCount: c.Get("in_flight_count").MustInt64(),
DeferredCount: c.Get("deferred_count").MustInt64(),
MessageCount: c.Get("message_count").MustInt64(),
RequeueCount: c.Get("requeue_count").MustInt64(),
TimeoutCount: c.Get("timeout_count").MustInt64(),
E2eProcessingLatency: e2eProcessingLatency,
// TODO: this is sort of wrong; clients should be de-duped
// client A that connects to NSQD-a and NSQD-b should only be counted once. right?
ClientCount: len(clients),
//.........这里部分代码省略.........
开发者ID:johntdyer,项目名称:golang-devops-stuff,代码行数:101,代码来源:lookupd.go
注:本文中的github.com/bitly/nsq/util.ApiRequest函数示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论