New Admin Info (#8497)

pkg/event/target/amqp.go

@@ -121,6 +121,18 @@ func (target *AMQPTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - Return true if target is up and active
+func (target *AMQPTarget) IsActive() (bool, error) {
+	ch, err := target.channel()
+	if err != nil {
+		return false, err
+	}
+	defer func() {
+		ch.Close()
+	}()
+	return true, nil
+}
+
 func (target *AMQPTarget) channel() (*amqp.Channel, error) {
 	var err error
 	var conn *amqp.Connection

pkg/event/target/elasticsearch.go

@@ -93,16 +93,25 @@ func (target *ElasticsearchTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - Return true if target is up and active
+func (target *ElasticsearchTarget) IsActive() (bool, error) {
+	if dErr := target.args.URL.DialHTTP(); dErr != nil {
+		if xnet.IsNetworkOrHostDown(dErr) {
+			return false, errNotConnected
+		}
+		return false, dErr
+	}
+	return true, nil
+}
+
 // Save - saves the events to the store if queuestore is configured, which will be replayed when the elasticsearch connection is active.
 func (target *ElasticsearchTarget) Save(eventData event.Event) error {
 	if target.store != nil {
 		return target.store.Put(eventData)
 	}
-	if dErr := target.args.URL.DialHTTP(); dErr != nil {
-		if xnet.IsNetworkOrHostDown(dErr) {
-			return errNotConnected
-		}
-		return dErr
+	_, err := target.IsActive()
+	if err != nil {
+		return err
 	}
 	return target.send(eventData)
 }
@@ -167,12 +176,9 @@ func (target *ElasticsearchTarget) Send(eventKey string) error {
 			return err
 		}
 	}
-
-	if dErr := target.args.URL.DialHTTP(); dErr != nil {
-		if xnet.IsNetworkOrHostDown(dErr) {
-			return errNotConnected
-		}
-		return dErr
+	_, err = target.IsActive()
+	if err != nil {
+		return err
 	}
 
 	eventData, eErr := target.store.Get(eventKey)

pkg/event/target/httpclient.go

@@ -44,6 +44,11 @@ func (target HTTPClientTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - does nothing and available for interface compatibility.
+func (target *HTTPClientTarget) IsActive() (bool, error) {
+	return true, nil
+}
+
 func (target *HTTPClientTarget) start() {
 	go func() {
 		defer func() {

pkg/event/target/kafka.go

@@ -124,13 +124,22 @@ func (target *KafkaTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - Return true if target is up and active
+func (target *KafkaTarget) IsActive() (bool, error) {
+	if !target.args.pingBrokers() {
+		return false, errNotConnected
+	}
+	return true, nil
+}
+
 // Save - saves the events to the store which will be replayed when the Kafka connection is active.
 func (target *KafkaTarget) Save(eventData event.Event) error {
 	if target.store != nil {
 		return target.store.Put(eventData)
 	}
-	if !target.args.pingBrokers() {
-		return errNotConnected
+	_, err := target.IsActive()
+	if err != nil {
+		return err
 	}
 	return target.send(eventData)
 }
@@ -162,9 +171,9 @@ func (target *KafkaTarget) send(eventData event.Event) error {
 // Send - reads an event from store and sends it to Kafka.
 func (target *KafkaTarget) Send(eventKey string) error {
 	var err error
-
-	if !target.args.pingBrokers() {
-		return errNotConnected
+	_, err = target.IsActive()
+	if err != nil {
+		return err
 	}
 
 	if target.producer == nil {

pkg/event/target/mqtt.go

@@ -121,6 +121,14 @@ func (target *MQTTTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - Return true if target is up and active
+func (target *MQTTTarget) IsActive() (bool, error) {
+	if !target.client.IsConnectionOpen() {
+		return false, errNotConnected
+	}
+	return true, nil
+}
+
 // send - sends an event to the mqtt.
 func (target *MQTTTarget) send(eventData event.Event) error {
 	objectName, err := url.QueryUnescape(eventData.S3.Object.Key)
@@ -144,8 +152,9 @@ func (target *MQTTTarget) send(eventData event.Event) error {
 // Send - reads an event from store and sends it to MQTT.
 func (target *MQTTTarget) Send(eventKey string) error {
 	// Do not send if the connection is not active.
-	if !target.client.IsConnectionOpen() {
-		return errNotConnected
+	_, err := target.IsActive()
+	if err != nil {
+		return err
 	}
 
 	eventData, err := target.store.Get(eventKey)
@@ -174,8 +183,9 @@ func (target *MQTTTarget) Save(eventData event.Event) error {
 	}
 
 	// Do not send if the connection is not active.
-	if !target.client.IsConnectionOpen() {
-		return errNotConnected
+	_, err := target.IsActive()
+	if err != nil {
+		return err
 	}
 
 	return target.send(eventData)

pkg/event/target/mysql.go

@@ -185,15 +185,24 @@ func (target *MySQLTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - Return true if target is up and active
+func (target *MySQLTarget) IsActive() (bool, error) {
+	if err := target.db.Ping(); err != nil {
+		if IsConnErr(err) {
+			return false, errNotConnected
+		}
+		return false, err
+	}
+	return true, nil
+}
+
 // Save - saves the events to the store which will be replayed when the SQL connection is active.
 func (target *MySQLTarget) Save(eventData event.Event) error {
 	if target.store != nil {
 		return target.store.Put(eventData)
 	}
-	if err := target.db.Ping(); err != nil {
-		if IsConnErr(err) {
-			return errNotConnected
-		}
+	_, err := target.IsActive()
+	if err != nil {
 		return err
 	}
 	return target.send(eventData)
@@ -244,10 +253,8 @@ func (target *MySQLTarget) send(eventData event.Event) error {
 // Send - reads an event from store and sends it to MySQL.
 func (target *MySQLTarget) Send(eventKey string) error {
 
-	if err := target.db.Ping(); err != nil {
-		if IsConnErr(err) {
-			return errNotConnected
-		}
+	_, err := target.IsActive()
+	if err != nil {
 		return err
 	}
 

pkg/event/target/nats.go

@@ -197,19 +197,28 @@ func (target *NATSTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - Return true if target is up and active
+func (target *NATSTarget) IsActive() (bool, error) {
+	if target.args.Streaming.Enable {
+		if !target.stanConn.NatsConn().IsConnected() {
+			return false, errNotConnected
+		}
+	} else {
+		if !target.natsConn.IsConnected() {
+			return false, errNotConnected
+		}
+	}
+	return true, nil
+}
+
 // Save - saves the events to the store which will be replayed when the Nats connection is active.
 func (target *NATSTarget) Save(eventData event.Event) error {
 	if target.store != nil {
 		return target.store.Put(eventData)
 	}
-	if target.args.Streaming.Enable {
-		if !target.stanConn.NatsConn().IsConnected() {
-			return errNotConnected
-		}
-	} else {
-		if !target.natsConn.IsConnected() {
-			return errNotConnected
-		}
+	_, err := target.IsActive()
+	if err != nil {
+		return err
 	}
 	return target.send(eventData)
 }

pkg/event/target/nsq.go

@@ -100,16 +100,25 @@ func (target *NSQTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - Return true if target is up and active
+func (target *NSQTarget) IsActive() (bool, error) {
+	if err := target.producer.Ping(); err != nil {
+		// To treat "connection refused" errors as errNotConnected.
+		if IsConnRefusedErr(err) {
+			return false, errNotConnected
+		}
+		return false, err
+	}
+	return true, nil
+}
+
 // Save - saves the events to the store which will be replayed when the nsq connection is active.
 func (target *NSQTarget) Save(eventData event.Event) error {
 	if target.store != nil {
 		return target.store.Put(eventData)
 	}
-	if err := target.producer.Ping(); err != nil {
-		// To treat "connection refused" errors as errNotConnected.
-		if IsConnRefusedErr(err) {
-			return errNotConnected
-		}
+	_, err := target.IsActive()
+	if err != nil {
 		return err
 	}
 	return target.send(eventData)
@@ -133,12 +142,8 @@ func (target *NSQTarget) send(eventData event.Event) error {
 
 // Send - reads an event from store and sends it to NSQ.
 func (target *NSQTarget) Send(eventKey string) error {
-
-	if err := target.producer.Ping(); err != nil {
-		// To treat "connection refused" errors as errNotConnected.
-		if IsConnRefusedErr(err) {
-			return errNotConnected
-		}
+	_, err := target.IsActive()
+	if err != nil {
 		return err
 	}
 

pkg/event/target/postgresql.go

@@ -183,15 +183,24 @@ func (target *PostgreSQLTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - Return true if target is up and active
+func (target *PostgreSQLTarget) IsActive() (bool, error) {
+	if err := target.db.Ping(); err != nil {
+		if IsConnErr(err) {
+			return false, errNotConnected
+		}
+		return false, err
+	}
+	return true, nil
+}
+
 // Save - saves the events to the store if questore is configured, which will be replayed when the PostgreSQL connection is active.
 func (target *PostgreSQLTarget) Save(eventData event.Event) error {
 	if target.store != nil {
 		return target.store.Put(eventData)
 	}
-	if err := target.db.Ping(); err != nil {
-		if IsConnErr(err) {
-			return errNotConnected
-		}
+	_, err := target.IsActive()
+	if err != nil {
 		return err
 	}
 	return target.send(eventData)
@@ -245,14 +254,10 @@ func (target *PostgreSQLTarget) send(eventData event.Event) error {
 
 // Send - reads an event from store and sends it to PostgreSQL.
 func (target *PostgreSQLTarget) Send(eventKey string) error {
-
-	if err := target.db.Ping(); err != nil {
-		if IsConnErr(err) {
-			return errNotConnected
-		}
+	_, err := target.IsActive()
+	if err != nil {
 		return err
 	}
-
 	if !target.firstPing {
 		if err := target.executeStmts(); err != nil {
 			if IsConnErr(err) {

pkg/event/target/redis.go

@@ -125,11 +125,8 @@ func (target *RedisTarget) ID() event.TargetID {
 	return target.id
 }
 
-// Save - saves the events to the store if questore is configured, which will be replayed when the redis connection is active.
-func (target *RedisTarget) Save(eventData event.Event) error {
-	if target.store != nil {
-		return target.store.Put(eventData)
-	}
+// IsActive - Return true if target is up and active
+func (target *RedisTarget) IsActive() (bool, error) {
 	conn := target.pool.Get()
 	defer func() {
 		cErr := conn.Close()
@@ -138,9 +135,21 @@ func (target *RedisTarget) Save(eventData event.Event) error {
 	_, pingErr := conn.Do("PING")
 	if pingErr != nil {
 		if IsConnRefusedErr(pingErr) {
-			return errNotConnected
+			return false, errNotConnected
 		}
-		return pingErr
+		return false, pingErr
+	}
+	return true, nil
+}
+
+// Save - saves the events to the store if questore is configured, which will be replayed when the redis connection is active.
+func (target *RedisTarget) Save(eventData event.Event) error {
+	if target.store != nil {
+		return target.store.Put(eventData)
+	}
+	_, err := target.IsActive()
+	if err != nil {
+		return err
 	}
 	return target.send(eventData)
 }

pkg/event/target/webhook.go

@@ -93,20 +93,29 @@ func (target WebhookTarget) ID() event.TargetID {
 	return target.id
 }
 
+// IsActive - Return true if target is up and active
+func (target *WebhookTarget) IsActive() (bool, error) {
+	u, pErr := xnet.ParseHTTPURL(target.args.Endpoint.String())
+	if pErr != nil {
+		return false, pErr
+	}
+	if dErr := u.DialHTTP(); dErr != nil {
+		if xnet.IsNetworkOrHostDown(dErr) {
+			return false, errNotConnected
+		}
+		return false, dErr
+	}
+	return true, nil
+}
+
 // Save - saves the events to the store if queuestore is configured, which will be replayed when the wenhook connection is active.
 func (target *WebhookTarget) Save(eventData event.Event) error {
 	if target.store != nil {
 		return target.store.Put(eventData)
 	}
-	u, pErr := xnet.ParseHTTPURL(target.args.Endpoint.String())
-	if pErr != nil {
-		return pErr
-	}
-	if dErr := u.DialHTTP(); dErr != nil {
-		if xnet.IsNetworkOrHostDown(dErr) {
-			return errNotConnected
-		}
-		return dErr
+	_, err := target.IsActive()
+	if err != nil {
+		return err
 	}
 	return target.send(eventData)
 }
@@ -153,17 +162,10 @@ func (target *WebhookTarget) send(eventData event.Event) error {
 
 // Send - reads an event from store and sends it to webhook.
 func (target *WebhookTarget) Send(eventKey string) error {
-	u, pErr := xnet.ParseHTTPURL(target.args.Endpoint.String())
-	if pErr != nil {
-		return pErr
+	_, err := target.IsActive()
+	if err != nil {
+		return err
 	}
-	if dErr := u.DialHTTP(); dErr != nil {
-		if xnet.IsNetworkOrHostDown(dErr) {
-			return errNotConnected
-		}
-		return dErr
-	}
-
 	eventData, eErr := target.store.Get(eventKey)
 	if eErr != nil {
 		// The last event key in a successful batch will be sent in the channel atmost once by the replayEvents()

pkg/event/targetlist.go

@@ -24,6 +24,7 @@ import (
 // Target - event target interface
 type Target interface {
 	ID() TargetID
+	IsActive() (bool, error)
 	Save(Event) error
 	Send(string) error
 	Close() error
@@ -130,6 +131,13 @@ func (list *TargetList) List() []TargetID {
 	return keys
 }
 
+// TargetMap - returns available targets.
+func (list *TargetList) TargetMap() map[TargetID]Target {
+	list.RLock()
+	defer list.RUnlock()
+	return list.targets
+}
+
 // Send - sends events to targets identified by target IDs.
 func (list *TargetList) Send(event Event, targetIDs ...TargetID) <-chan TargetIDErr {
 	errCh := make(chan TargetIDErr)

pkg/event/targetlist_test.go

@@ -67,6 +67,10 @@ func (target ExampleTarget) Close() error {
 	return nil
 }
 
+func (target ExampleTarget) IsActive() (bool, error) {
+	return false, errors.New("not connected to target server/service")
+}
+
 func TestTargetListAdd(t *testing.T) {
 	targetListCase1 := NewTargetList()