Merge pull request #134 from kaleido-io/add-monitoring-routes

add monitoring routes

Author: Chengxuan

cmd/client_eventstreams_delete.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -30,7 +30,7 @@ func clientEventStreamsDeleteCommand(clientFactory func() (apiclient.FFTMClient,
 		Use:   "delete",
 		Short: "Delete event streams",
 		Long:  "",
-		RunE: func(cmd *cobra.Command, args []string) error {
+		RunE: func(_ *cobra.Command, _ []string) error {
 			client, err := clientFactory()
 			if err != nil {
 				return err

cmd/client_eventstreams_list.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -30,7 +30,7 @@ func clientEventStreamsListCommand(clientFactory func() (apiclient.FFTMClient, e
 		Use:   "list",
 		Short: "List event streams",
 		Long:  "",
-		RunE: func(cmd *cobra.Command, args []string) error {
+		RunE: func(_ *cobra.Command, _ []string) error {
 			client, err := clientFactory()
 			if err != nil {
 				return err

cmd/client_listeners_delete.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -30,7 +30,7 @@ func clientListenersDeleteCommand(clientFactory func() (apiclient.FFTMClient, er
 		Use:   "delete",
 		Short: "Delete event streams",
 		Long:  "",
-		RunE: func(cmd *cobra.Command, args []string) error {
+		RunE: func(_ *cobra.Command, _ []string) error {
 			client, err := clientFactory()
 			if err != nil {
 				return err

cmd/client_listeners_list.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -30,7 +30,7 @@ func clientListenersListCommand(clientFactory func() (apiclient.FFTMClient, erro
 		Use:   "list",
 		Short: "List listeners",
 		Long:  "",
-		RunE: func(cmd *cobra.Command, args []string) error {
+		RunE: func(_ *cobra.Command, _ []string) error {
 			client, err := clientFactory()
 			if err != nil {
 				return err

cmd/migrate.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -41,7 +41,7 @@ func buildLeveldb2postgresCommand(initConfig func() error) *cobra.Command {
 	leveldb2postgresEventStreamsCmd := &cobra.Command{
 		Use:   "leveldb2postgres",
 		Short: "Migrate from LevelDB to PostgreSQL persistence",
-		RunE: func(cmd *cobra.Command, args []string) error {
+		RunE: func(_ *cobra.Command, _ []string) error {
 			if err := initConfig(); err != nil {
 				return err
 			}

config.md

@@ -175,8 +175,8 @@
 |Key|Description|Type|Default Value|
 |---|-----------|----|-------------|
 |address|The IP address on which the metrics HTTP API should listen|`int`|`127.0.0.1`
-|enabled|Enables the metrics API|`boolean`|`false`
-|path|The path from which to serve the Prometheus metrics|`string`|`/metrics`
+|enabled|Deprecated: Please use 'monitoring.enabled' instead|`boolean`|`false`
+|path|Deprecated: Please use 'monitoring.metricsPath' instead|`string`|`/metrics`
 |port|The port on which the metrics HTTP API should listen|`int`|`6000`
 |publicURL|The fully qualified public URL for the metrics API. This is used for building URLs in HTTP responses and in OpenAPI Spec generation|URL `string`|`<nil>`
 |readTimeout|The maximum time to wait when reading from an HTTP connection|[`time.Duration`](https://pkg.go.dev/time#Duration)|`15s`
@@ -210,6 +210,46 @@
 |keyFile|The path to the private key file for TLS on this API|`string`|`<nil>`
 |requiredDNAttributes|A set of required subject DN attributes. Each entry is a regular expression, and the subject certificate must have a matching attribute of the specified type (CN, C, O, OU, ST, L, STREET, POSTALCODE, SERIALNUMBER are valid attributes)|`map[string]string`|`<nil>`
 
+## monitoring
+
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|address|Listener address|`int`|`127.0.0.1`
+|enabled|Enables the monitoring APIs|`boolean`|`false`
+|metricsPath|The path from which to serve the Prometheus metrics|`string`|`/metrics`
+|port|Listener port|`int`|`6000`
+|publicURL|Externally available URL for the HTTP endpoint|`string`|`<nil>`
+|readTimeout|HTTP server read timeout|[`time.Duration`](https://pkg.go.dev/time#Duration)|`15s`
+|shutdownTimeout|HTTP server shutdown timeout|[`time.Duration`](https://pkg.go.dev/time#Duration)|`10s`
+|writeTimeout|HTTP server write timeout|[`time.Duration`](https://pkg.go.dev/time#Duration)|`15s`
+
+## monitoring.auth
+
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|type|The auth plugin to use for server side authentication of requests|`string`|`<nil>`
+
+## monitoring.auth.basic
+
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|passwordfile|The path to a .htpasswd file to use for authenticating requests. Passwords should be hashed with bcrypt.|`string`|`<nil>`
+
+## monitoring.tls
+
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|ca|The TLS certificate authority in PEM format (this option is ignored if caFile is also set)|`string`|`<nil>`
+|caFile|The path to the CA file for TLS on this API|`string`|`<nil>`
+|cert|The TLS certificate in PEM format (this option is ignored if certFile is also set)|`string`|`<nil>`
+|certFile|The path to the certificate file for TLS on this API|`string`|`<nil>`
+|clientAuth|Enables or disables client auth for TLS on this API|`string`|`<nil>`
+|enabled|Enables or disables TLS on this API|`boolean`|`false`
+|insecureSkipHostVerify|When to true in unit test development environments to disable TLS verification. Use with extreme caution|`boolean`|`<nil>`
+|key|The TLS certificate key in PEM format (this option is ignored if keyFile is also set)|`string`|`<nil>`
+|keyFile|The path to the private key file for TLS on this API|`string`|`<nil>`
+|requiredDNAttributes|A set of required subject DN attributes. Each entry is a regular expression, and the subject certificate must have a matching attribute of the specified type (CN, C, O, OU, ST, L, STREET, POSTALCODE, SERIALNUMBER are valid attributes)|`map[string]string`|`<nil>`
+
 ## persistence
 
 |Key|Description|Type|Default Value|

internal/apiclient/eventstreams.go

@@ -1,4 +1,4 @@
-// Copyright © 2023 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -18,6 +18,7 @@ package apiclient
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"regexp"
 
@@ -31,7 +32,7 @@ func (c *fftmClient) GetEventStreams(ctx context.Context) ([]apitypes.EventStrea
 		SetResult(&eventStreams).
 		Get("eventstreams")
 	if !resp.IsSuccess() {
-		return nil, fmt.Errorf(string(resp.Body()))
+		return nil, errors.New(string(resp.Body()))
 	}
 	return eventStreams, err
 }
@@ -43,7 +44,7 @@ func (c *fftmClient) GetListeners(ctx context.Context, eventStreamID string) ([]
 		SetResult(&listeners).
 		Get(fmt.Sprintf("eventstreams/%s/listeners", eventStreamID))
 	if !resp.IsSuccess() {
-		return nil, fmt.Errorf(string(resp.Body()))
+		return nil, errors.New(string(resp.Body()))
 	}
 	return listeners, err
 }
@@ -56,7 +57,7 @@ func (c *fftmClient) DeleteEventStream(ctx context.Context, eventStreamID string
 		return err
 	}
 	if !resp.IsSuccess() {
-		return fmt.Errorf(string(resp.Body()))
+		return errors.New(string(resp.Body()))
 	}
 	return nil
 }
@@ -92,7 +93,7 @@ func (c *fftmClient) DeleteListener(ctx context.Context, eventStreamID, listener
 		return err
 	}
 	if !resp.IsSuccess() {
-		return fmt.Errorf(string(resp.Body()))
+		return errors.New(string(resp.Body()))
 	}
 	return nil
 }

internal/confirmations/confirmations.go

@@ -1,4 +1,4 @@
-// Copyright © 2024 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -311,6 +311,7 @@ func (bcm *blockConfirmationManager) getBlockByHash(blockHash string) (*apitypes
 
 func (bcm *blockConfirmationManager) getBlockByNumber(blockNumber uint64, allowCache bool, expectedParentHash string) (*apitypes.BlockInfo, error) {
 	res, reason, err := bcm.connector.BlockInfoByNumber(bcm.ctx, &ffcapi.BlockInfoByNumberRequest{
+		//nolint:gosec
 		BlockNumber:        fftypes.NewFFBigInt(int64(blockNumber)),
 		AllowCache:         allowCache,
 		ExpectedParentHash: expectedParentHash,

internal/confirmations/confirmed_block_listener.go

@@ -1,4 +1,4 @@
-// Copyright © 2024 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -357,6 +357,7 @@ func (cbl *confirmedBlockListener) dispatchAllConfirmed() {
 				BlockEvent: &ffcapi.BlockEvent{
 					ListenerID: cbl.id,
 					BlockInfo: ffcapi.BlockInfo{
+						//nolint:gosec
 						BlockNumber:       fftypes.NewFFBigInt(int64(block.BlockNumber)),
 						BlockHash:         block.BlockHash,
 						ParentHash:        block.ParentHash,

internal/events/eventstream.go

@@ -1,4 +1,4 @@
-// Copyright © 2024 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -65,6 +65,7 @@ var esDefaults struct {
 }
 
 func InitDefaults() {
+	//nolint:gosec
 	esDefaults.batchSize = config.GetInt64(tmconfig.EventStreamsDefaultsBatchSize)
 	esDefaults.batchTimeout = fftypes.FFDuration(config.GetDuration(tmconfig.EventStreamsDefaultsBatchTimeout))
 	esDefaults.errorHandling = fftypes.FFEnum(config.GetString(tmconfig.EventStreamsDefaultsErrorHandling))
@@ -218,6 +219,7 @@ func mergeValidateEsConfig(ctx context.Context, base *apitypes.EventStream, upda
 
 	// Batch timeout
 	if updates.EthCompatBatchTimeoutMS != nil {
+		//nolint:gosec
 		dv := fftypes.FFDuration(*updates.EthCompatBatchTimeoutMS) * fftypes.FFDuration(time.Millisecond)
 		changed = apitypes.CheckUpdateDuration(changed, &merged.BatchTimeout, base.BatchTimeout, &dv, esDefaults.batchTimeout)
 	} else {
@@ -226,6 +228,7 @@ func mergeValidateEsConfig(ctx context.Context, base *apitypes.EventStream, upda
 
 	// Retry timeout
 	if updates.EthCompatRetryTimeoutSec != nil {
+		//nolint:gosec
 		dv := fftypes.FFDuration(*updates.EthCompatRetryTimeoutSec) * fftypes.FFDuration(time.Second)
 		changed = apitypes.CheckUpdateDuration(changed, &merged.RetryTimeout, base.RetryTimeout, &dv, esDefaults.retryTimeout)
 	} else {
@@ -234,6 +237,7 @@ func mergeValidateEsConfig(ctx context.Context, base *apitypes.EventStream, upda
 
 	// Blocked retry delay
 	if updates.EthCompatBlockedRetryDelaySec != nil {
+		//nolint:gosec
 		dv := fftypes.FFDuration(*updates.EthCompatBlockedRetryDelaySec) * fftypes.FFDuration(time.Second)
 		changed = apitypes.CheckUpdateDuration(changed, &merged.BlockedRetryDelay, base.BlockedRetryDelay, &dv, esDefaults.blockedRetryDelay)
 	} else {
@@ -509,7 +513,8 @@ func (es *eventStream) Start(ctx context.Context) error {
 		startTime:     fftypes.Now(),
 		eventLoopDone: make(chan struct{}),
 		batchLoopDone: make(chan struct{}),
-		updates:       make(chan *ffcapi.ListenerEvent, int(*es.spec.BatchSize)),
+		//nolint:gosec
+		updates: make(chan *ffcapi.ListenerEvent, int(*es.spec.BatchSize)),
 	}
 	startedState.ctx, startedState.cancelCtx = context.WithCancel(es.bgCtx)
 	es.currentState = startedState
@@ -781,6 +786,8 @@ func (es *eventStream) checkConfirmedEventForBatch(e *ffcapi.ListenerEvent) (l *
 func (es *eventStream) batchLoop(startedState *startedStreamState) {
 	defer close(startedState.batchLoopDone)
 	ctx := startedState.ctx
+
+	//nolint:gosec
 	maxSize := int(*es.spec.BatchSize)
 	batchNumber := int64(0)
 
@@ -955,7 +962,7 @@ func (es *eventStream) writeCheckpoint(startedState *startedStreamState, batch *
 	}
 
 	// We only return if the context is cancelled, or the checkpoint succeeds
-	return es.retry.Do(startedState.ctx, "checkpoint", func(attempt int) (retry bool, err error) {
+	return es.retry.Do(startedState.ctx, "checkpoint", func(_ int) (retry bool, err error) {
 		return true, es.persistence.WriteCheckpoint(startedState.ctx, cp)
 	})
 }