<commit_before>/* Copyright 2015 LinkedIn Corp. Licensed under the Apache License, Version
 * 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 */

package main

import (
	"encoding/json"
	"errors"
	log "github.com/cihub/seelog"
	"github.com/samuel/go-zookeeper/zk"
	"math/rand"
	"regexp"
	"strconv"
	"sync"
	"time"
)

type StormClient struct {
	app                *ApplicationContext
	cluster            string
	conn               *zk.Conn
	stormRefreshTicker *time.Ticker
	stormGroupList     map[string]bool
	stormGroupLock     sync.RWMutex
}

type Topology struct {
	Id   string
	Name string
}

type Broker struct {
	Host string
	Port int
}

type SpoutState struct {
	Topology  Topology
	Offset    int
	Partition int
	Broker    Broker
	Topic     string
}

func NewStormClient(app *ApplicationContext, cluster string) (*StormClient, error) {
	// here we share the timeout w/ global zk
	zkconn, _, err := zk.Connect(app.Config.Storm[cluster].Zookeepers, time.Duration(app.Config.Zookeeper.Timeout)*time.Second)
	if err != nil {
		return nil, err
	}

	client := &StormClient{
		app:            app,
		cluster:        cluster,
		conn:           zkconn,
		stormGroupLock: sync.RWMutex{},
		stormGroupList: make(map[string]bool),
	}

	// Now get the first set of offsets and start a goroutine to continually check them
	client.refreshConsumerGroups()
	client.stormRefreshTicker = time.NewTicker(time.Duration(client.app.Config.Lagcheck.StormGroupRefresh) * time.Second)
	go func() {
		for _ = range client.stormRefreshTicker.C {
			client.refreshConsumerGroups()
		}
	}()

	return client, nil
}

func (stormClient *StormClient) Stop() {

	if stormClient.stormRefreshTicker != nil {
		stormClient.stormRefreshTicker.Stop()
		stormClient.stormGroupLock.Lock()
		stormClient.stormGroupList = make(map[string]bool)
		stormClient.stormGroupLock.Unlock()
	}

	stormClient.conn.Close()
}

func parsePartitionId(partitionStr string) (int, error) {
	re := regexp.MustCompile(`^partition_([0-9]+)$`)
	if parsed := re.FindStringSubmatch(partitionStr); len(parsed) == 2 {
		return strconv.Atoi(parsed[1])
	} else {
		return -1, errors.New("Invalid partition id: " + partitionStr)
	}
}

func parseStormSpoutStateJson(stateStr string) (int, string, error) {
	stateJson := new(SpoutState)
	if err := json.Unmarshal([]byte(stateStr), &stateJson); err == nil {
		log.Debugf("Parsed storm state [%s] to structure [%v]", stateStr, stateJson)
		return stateJson.Offset, stateJson.Topic, nil
	} else {
		return -1, "", err
	}
}

func (stormClient *StormClient) getConsumerGroupPath(consumerGroup string) string {
	if "/" == stormClient.app.Config.Storm[stormClient.cluster].ZookeeperPath {
		return "/" + consumerGroup
	} else {
		return stormClient.app.Config.Storm[stormClient.cluster].ZookeeperPath + "/" + consumerGroup
	}
}

func (stormClient *StormClient) getOffsetsForConsumerGroup(consumerGroup string) {
	consumerGroupPath := getConsumerGroupPath(consumerGroup)
	partition_ids, _, err := stormClient.conn.Children(consumerGroupPath)
	switch {
	case err == nil:
		for _, partition_id := range partition_ids {
			partition, errConversion := parsePartitionId(partition_id)
			switch {
			case errConversion == nil:
				stormClient.getOffsetsForPartition(consumerGroup, partition, consumerGroupPath+"/"+partition_id)
			default:<commit_msg>Fix function call issue in storm<commit_after>/* Copyright 2015 LinkedIn Corp. Licensed under the Apache License, Version
 * 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 */

package main

import (
	"encoding/json"
	"errors"
	log "github.com/cihub/seelog"
	"github.com/samuel/go-zookeeper/zk"
	"math/rand"
	"regexp"
	"strconv"
	"sync"
	"time"
)

type StormClient struct {
	app                *ApplicationContext
	cluster            string
	conn               *zk.Conn
	stormRefreshTicker *time.Ticker
	stormGroupList     map[string]bool
	stormGroupLock     sync.RWMutex
}

type Topology struct {
	Id   string
	Name string
}

type Broker struct {
	Host string
	Port int
}

type SpoutState struct {
	Topology  Topology
	Offset    int
	Partition int
	Broker    Broker
	Topic     string
}

func NewStormClient(app *ApplicationContext, cluster string) (*StormClient, error) {
	// here we share the timeout w/ global zk
	zkconn, _, err := zk.Connect(app.Config.Storm[cluster].Zookeepers, time.Duration(app.Config.Zookeeper.Timeout)*time.Second)
	if err != nil {
		return nil, err
	}

	client := &StormClient{
		app:            app,
		cluster:        cluster,
		conn:           zkconn,
		stormGroupLock: sync.RWMutex{},
		stormGroupList: make(map[string]bool),
	}

	// Now get the first set of offsets and start a goroutine to continually check them
	client.refreshConsumerGroups()
	client.stormRefreshTicker = time.NewTicker(time.Duration(client.app.Config.Lagcheck.StormGroupRefresh) * time.Second)
	go func() {
		for _ = range client.stormRefreshTicker.C {
			client.refreshConsumerGroups()
		}
	}()

	return client, nil
}

func (stormClient *StormClient) Stop() {

	if stormClient.stormRefreshTicker != nil {
		stormClient.stormRefreshTicker.Stop()
		stormClient.stormGroupLock.Lock()
		stormClient.stormGroupList = make(map[string]bool)
		stormClient.stormGroupLock.Unlock()
	}

	stormClient.conn.Close()
}

func parsePartitionId(partitionStr string) (int, error) {
	re := regexp.MustCompile(`^partition_([0-9]+)$`)
	if parsed := re.FindStringSubmatch(partitionStr); len(parsed) == 2 {
		return strconv.Atoi(parsed[1])
	} else {
		return -1, errors.New("Invalid partition id: " + partitionStr)
	}
}

func parseStormSpoutStateJson(stateStr string) (int, string, error) {
	stateJson := new(SpoutState)
	if err := json.Unmarshal([]byte(stateStr), &stateJson); err == nil {
		log.Debugf("Parsed storm state [%s] to structure [%v]", stateStr, stateJson)
		return stateJson.Offset, stateJson.Topic, nil
	} else {
		return -1, "", err
	}
}

func (stormClient *StormClient) getConsumerGroupPath(consumerGroup string) string {
	if "/" == stormClient.app.Config.Storm[stormClient.cluster].ZookeeperPath {
		return "/" + consumerGroup
	} else {
		return stormClient.app.Config.Storm[stormClient.cluster].ZookeeperPath + "/" + consumerGroup
	}
}

func (stormClient *StormClient) getOffsetsForConsumerGroup(consumerGroup string) {
	consumerGroupPath := stormClient.getConsumerGroupPath(consumerGroup)
	partition_ids, _, err := stormClient.conn.Children(consumerGroupPath)
	switch {
	case err == nil:
		for _, partition_id := range partition_ids {
			partition, errConversion := parsePartitionId(partition_id)
			switch {
			case errConversion == nil:
				stormClient.getOffsetsForPartition(consumerGroup, partition, consumerGroupPath+"/"+partition_id)
			default:
{'fraction_non_alphanumeric': 0.10878717913893793, 'fraction_numerical': 0.003802797772647019, 'mean_word_length': 4.204240282685513, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '270822', 'n_tokens_mistral': 2431, 'n_tokens_neox': 2326, 'n_words': 598}