<commit_before>// Copyright (c) 2017 Cisco and/or its affiliates.
//
// 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.
// See the License for the specific language governing permissions and
// limitations under the License.

package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/ligato/cn-infra/db/keyval"
	"github.com/ligato/cn-infra/flavors/local"
	"github.com/ligato/cn-infra/health/statuscheck"
	"github.com/ligato/cn-infra/logging"
	"github.com/ligato/cn-infra/messaging"
	"github.com/ligato/cn-infra/messaging/kafka/client"
	"github.com/ligato/cn-infra/messaging/kafka/mux"
	"github.com/ligato/cn-infra/utils/safeclose"
)

const topic = "status-check"

// Plugin provides API for interaction with kafka brokers.
type Plugin struct {
	Deps         // inject
	mux          *mux.Multiplexer
	subscription chan (*client.ConsumerMessage)

	// Kafka plugin is using two clients. The first one is using 'hash' (default) partitioner. The second mux
	// uses manual partitioner which allows to send a message to specified partition and watching to desired partition/offset
	hsClient  sarama.Client
	manClient sarama.Client

	disabled bool
}

// Deps is here to group injected dependencies of plugin
// to not mix with other plugin fields.
type Deps struct {
	local.PluginInfraDeps //inject
}

// FromExistingMux is used mainly for testing purposes.
func FromExistingMux(mux *mux.Multiplexer) *Plugin {
	return &Plugin{mux: mux}
}

// Init is called at plugin initialization.
func (plugin *Plugin) Init() (err error) {
	// Prepare topic and  subscription for status check client
	plugin.subscription = make(chan *client.ConsumerMessage)

	// Get muxCfg data (contains kafka brokers ip addresses)
	muxCfg := &mux.Config{}
	found, err := plugin.PluginConfig.GetValue(muxCfg)
	if !found {
		plugin.Log.Info("kafka config not found ", plugin.PluginConfig.GetConfigName(), " - skip loading this plugin")
		plugin.disabled = true
		return nil //skip loading the plugin
	}
	if err != nil {
		return err
	}
	// retrieve clientCfg
	clientCfg := plugin.getClientConfig(muxCfg, plugin.Log, topic)

	// init 'hash' sarama client
	plugin.hsClient, err = client.NewClient(clientCfg, client.Hash)
	if err != nil {
		return err
	}

	// init 'manual' sarama client
	plugin.manClient, err = client.NewClient(clientCfg, client.Manual)
	if err != nil {
		return err
	}

	// Initialize both multiplexers to allow both, dynamic and manual mode
	if plugin.mux == nil {
		name := plugin.ServiceLabel.GetAgentLabel() + "-hash"
		plugin.mux, err = mux.InitMultiplexerWithConfig(clientCfg, plugin.hsClient, plugin.manClient, name, plugin.Log)
		if err != nil {
			return err
		}
		plugin.Log.Debug("Default multiplexer initialized")
	}

	return err
}

// AfterInit is called in the second phase of the initialization. The kafka multiplexerNewWatcher
// is started, all consumers have to be subscribed until this phase.
func (plugin *Plugin) AfterInit() error {
	if plugin.mux != nil {
		err := plugin.mux.Start()
		if err != nil {
			return err
		}
	}

	// Register for providing status reports (polling mode)
	if plugin.StatusCheck != nil && !plugin.disabled {
		plugin.StatusCheck.Register(plugin.PluginName, func() (statuscheck.PluginState, error) {
			if plugin.mux.Consumer == nil || plugin.mux.Consumer.Client == nil {
				return statuscheck.Error, fmt.Errorf("kafka client/consumer not initialized")
			}
			// Method 'RefreshMetadata()' returns error if kafka server is unavailable<commit_msg>fixed kafka statuscheck<commit_after>// Copyright (c) 2017 Cisco and/or its affiliates.
//
// 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.
// See the License for the specific language governing permissions and
// limitations under the License.

package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/ligato/cn-infra/db/keyval"
	"github.com/ligato/cn-infra/flavors/local"
	"github.com/ligato/cn-infra/health/statuscheck"
	"github.com/ligato/cn-infra/logging"
	"github.com/ligato/cn-infra/messaging"
	"github.com/ligato/cn-infra/messaging/kafka/client"
	"github.com/ligato/cn-infra/messaging/kafka/mux"
	"github.com/ligato/cn-infra/utils/safeclose"
)

const topic = "status-check"

// Plugin provides API for interaction with kafka brokers.
type Plugin struct {
	Deps         // inject
	mux          *mux.Multiplexer
	subscription chan (*client.ConsumerMessage)

	// Kafka plugin is using two clients. The first one is using 'hash' (default) partitioner. The second mux
	// uses manual partitioner which allows to send a message to specified partition and watching to desired partition/offset
	hsClient  sarama.Client
	manClient sarama.Client

	disabled bool
}

// Deps is here to group injected dependencies of plugin
// to not mix with other plugin fields.
type Deps struct {
	local.PluginInfraDeps //inject
}

// FromExistingMux is used mainly for testing purposes.
func FromExistingMux(mux *mux.Multiplexer) *Plugin {
	return &Plugin{mux: mux}
}

// Init is called at plugin initialization.
func (plugin *Plugin) Init() (err error) {
	// Prepare topic and  subscription for status check client
	plugin.subscription = make(chan *client.ConsumerMessage)

	// Get muxCfg data (contains kafka brokers ip addresses)
	muxCfg := &mux.Config{}
	found, err := plugin.PluginConfig.GetValue(muxCfg)
	if !found {
		plugin.Log.Info("kafka config not found ", plugin.PluginConfig.GetConfigName(), " - skip loading this plugin")
		plugin.disabled = true
		return nil //skip loading the plugin
	}
	if err != nil {
		return err
	}
	// retrieve clientCfg
	clientCfg := plugin.getClientConfig(muxCfg, plugin.Log, topic)

	// init 'hash' sarama client
	plugin.hsClient, err = client.NewClient(clientCfg, client.Hash)
	if err != nil {
		return err
	}

	// init 'manual' sarama client
	plugin.manClient, err = client.NewClient(clientCfg, client.Manual)
	if err != nil {
		return err
	}

	// Initialize both multiplexers to allow both, dynamic and manual mode
	if plugin.mux == nil {
		name := plugin.ServiceLabel.GetAgentLabel() + "-hash"
		plugin.mux, err = mux.InitMultiplexerWithConfig(clientCfg, plugin.hsClient, plugin.manClient, name, plugin.Log)
		if err != nil {
			return err
		}
		plugin.Log.Debug("Default multiplexer initialized")
	}

	return err
}

// AfterInit is called in the second phase of the initialization. The kafka multiplexerNewWatcher
// is started, all consumers have to be subscribed until this phase.
func (plugin *Plugin) AfterInit() error {
	if plugin.mux != nil {
		err := plugin.mux.Start()
		if err != nil {
			return err
		}
	}

	// Register for providing status reports (polling mode)
	if plugin.StatusCheck != nil && !plugin.disabled {
		plugin.StatusCheck.Register(plugin.PluginName, func() (statuscheck.PluginState, error) {
			if plugin.hsClient == nil || plugin.hsClient.Closed() {
				return statuscheck.Error, fmt.Errorf("kafka client/consumer not initialized")
			}
			// Method 'RefreshMetadata()' returns error if kafka server is unavailable
{'fraction_non_alphanumeric': 0.10518194676610518, 'fraction_numerical': 0.0020573485920020573, 'mean_word_length': 3.804200123533045, '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': '6596098', 'n_tokens_mistral': 2453, 'n_tokens_neox': 2317, 'n_words': 788}