Local file storage extension

cmd/otelcontribcol/components.go

@@ -46,6 +46,7 @@ import (
  "github.com/open-telemetry/opentelemetry-collector-contrib/extension/httpforwarder"
  "github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer/hostobserver"
  "github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer/k8sobserver"
+ "github.com/open-telemetry/opentelemetry-collector-contrib/extension/storage/filestorage"
  "github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbyattrsprocessor"
  "github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbytraceprocessor"
  "github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sprocessor"
@@ -91,6 +92,7 @@ func components() (component.Factories, error) {
  hostobserver.NewFactory(),
  httpforwarder.NewFactory(),
  k8sobserver.NewFactory(),
+ filestorage.NewFactory(),
  }
 
  for _, ext := range factories.Extensions {

extension/storage/README.md

@@ -14,6 +14,3 @@ Set(string, []byte) error
 Delete(string) error
 ```
 Note: All methods should return error only if a problem occurred. (For example, if a file is no longer accessible, or if a remote service is unavailable.)
-
-# TODO Sample code
-- Document component expectations

extension/storage/filestorage/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

extension/storage/filestorage/README.md

@@ -0,0 +1,34 @@
+# File Storage
+
+The File Storage extension can persist state to the local file system. 
+
+The extension requires read and write access to a directory. A default directory can be used, but it must already exist in order for the extension to operate.
+
+`directory` is the relative or absolute path to the dedicated data storage directory. 
+
+`timeout` is the maximum time to wait for a file lock. This value does not need to be modified in most circumstances.
+
+
+```
+extensions:
+ file_storage:
+ file_storage/all_settings:
+ directory: /var/lib/otelcol/mydir
+ timeout: 1s
+
+service:
+ extensions: [file_storage, file_storage/all_settings]
+ pipelines:
+ traces:
+ receivers: [nop]
+ processors: [nop]
+ exporters: [nop]
+
+# Data pipeline is required to load the config.
+receivers:
+ nop:
+processors:
+ nop:
+exporters:
+ nop:
+```

extension/storage/filestorage/client.go

@@ -0,0 +1,99 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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
+//
+// https://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 filestorage
+
+import (
+ "context"
+ "errors"
+ "time"
+
+ "go.etcd.io/bbolt"
+)
+
+var defaultBucket = []byte(`default`)
+
+type fileStorageClient struct {
+ db *bbolt.DB
+}
+
+func newClient(filePath string, timeout time.Duration) (*fileStorageClient, error) {
+ options := &bbolt.Options{
+ Timeout: timeout,
+ NoSync: true,
+ }
+ db, err := bbolt.Open(filePath, 0600, options)
+ if err != nil {
+ return nil, err
+ }
+
+ initBucket := func(tx *bbolt.Tx) error {
+ _, err := tx.CreateBucketIfNotExists(defaultBucket)
+ return err
+ }
+ if err := db.Update(initBucket); err != nil {
+ return nil, err
+ }
+
+ return &fileStorageClient{db}, nil
+}
+
+// Get will retrieve data from storage that corresponds to the specified key
+func (c *fileStorageClient) Get(_ context.Context, key string) ([]byte, error) {
+ var result []byte
+ get := func(tx *bbolt.Tx) error {
+ bucket := tx.Bucket(defaultBucket)
+ if bucket == nil {
+ return errors.New("storage not initialized")
+ }
+ result = bucket.Get([]byte(key))
+ return nil // no error
+ }
+
+ if err := c.db.Update(get); err != nil {
+ return nil, err
+ }
+ return result, nil
+}
+
+// Set will store data. The data can be retrieved using the same key
+func (c *fileStorageClient) Set(_ context.Context, key string, value []byte) error {
+ set := func(tx *bbolt.Tx) error {
+ bucket := tx.Bucket(defaultBucket)
+ if bucket == nil {
+ return errors.New("storage not initialized")
+ }
+ return bucket.Put([]byte(key), value)
+ }
+
+ return c.db.Update(set)
+}
+
+// Delete will delete data associated with the specified key
+func (c *fileStorageClient) Delete(_ context.Context, key string) error {
+ delete := func(tx *bbolt.Tx) error {
+ bucket := tx.Bucket(defaultBucket)
+ if bucket == nil {
+ return errors.New("storage not initialized")
+ }
+ return bucket.Delete([]byte(key))
+ }
+
+ return c.db.Update(delete)
+}
+
+// Close will close the database
+func (c *fileStorageClient) close() error {
+ return c.db.Close()
+}

extension/storage/filestorage/client_test.go

@@ -0,0 +1,194 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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
+//
+// https://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 filestorage
+
+import (
+ "context"
+ "io/ioutil"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+ "go.etcd.io/bbolt"
+)
+
+func TestClientOperations(t *testing.T) {
+ tempDir := newTempDir(t)
+ dbFile := filepath.Join(tempDir, "my_db")
+
+ client, err := newClient(dbFile, time.Second)
+ require.NoError(t, err)
+
+ ctx := context.Background()
+ testKey := "testKey"
+ testValue := []byte("testValue")
+
+ // Make sure nothing is there
+ value, err := client.Get(ctx, testKey)
+ require.NoError(t, err)
+ require.Nil(t, value)
+
+ // Set it
+ err = client.Set(ctx, testKey, testValue)
+ require.NoError(t, err)
+
+ // Get it back out, make sure it's right
+ value, err = client.Get(ctx, testKey)
+ require.NoError(t, err)
+ require.Equal(t, testValue, value)
+
+ // Delete it
+ err = client.Delete(ctx, testKey)
+ require.NoError(t, err)
+
+ // Make sure it's gone
+ value, err = client.Get(ctx, testKey)
+ require.NoError(t, err)
+ require.Nil(t, value)
+}
+
+func TestNewClientTransactionErrors(t *testing.T) {
+ timeout := 100 * time.Millisecond
+
+ testKey := "testKey"
+ testValue := []byte("testValue")
+
+ testCases := []struct {
+ name string
+ setup func(*bbolt.Tx) error
+ validate func(*testing.T, *fileStorageClient)
+ }{
+ {
+ name: "get",
+ setup: func(tx *bbolt.Tx) error {
+ return tx.DeleteBucket(defaultBucket)
+ },
+ validate: func(t *testing.T, c *fileStorageClient) {
+ value, err := c.Get(context.Background(), testKey)
+ require.Error(t, err)
+ require.Equal(t, "storage not initialized", err.Error())
+ require.Nil(t, value)
+ },
+ },
+ {
+ name: "set",
+ setup: func(tx *bbolt.Tx) error {
+ return tx.DeleteBucket(defaultBucket)
+ },
+ validate: func(t *testing.T, c *fileStorageClient) {
+ err := c.Set(context.Background(), testKey, testValue)
+ require.Error(t, err)
+ require.Equal(t, "storage not initialized", err.Error())
+ },
+ },
+ {
+ name: "delete",
+ setup: func(tx *bbolt.Tx) error {
+ return tx.DeleteBucket(defaultBucket)
+ },
+ validate: func(t *testing.T, c *fileStorageClient) {
+ err := c.Delete(context.Background(), testKey)
+ require.Error(t, err)
+ require.Equal(t, "storage not initialized", err.Error())
+ },
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+
+ tempDir := newTempDir(t)
+ dbFile := filepath.Join(tempDir, "my_db")
+
+ client, err := newClient(dbFile, timeout)
+ require.NoError(t, err)
+
+ // Create a problem
+ client.db.Update(tc.setup)
+
+ // Validate expected behavior
+ tc.validate(t, client)
+ })
+ }
+}
+
+func TestNewClientErrorsOnInvalidBucket(t *testing.T) {
+ temp := defaultBucket
+ defaultBucket = nil
+
+ tempDir := newTempDir(t)
+ dbFile := filepath.Join(tempDir, "my_db")
+
+ client, err := newClient(dbFile, time.Second)
+ require.Error(t, err)
+ require.Nil(t, client)
+
+ defaultBucket = temp
+}
+
+func BenchmarkClientGet(b *testing.B) {
+ tempDir := newTempDir(b)
+ dbFile := filepath.Join(tempDir, "my_db")
+
+ client, err := newClient(dbFile, time.Second)
+ require.NoError(b, err)
+
+ ctx := context.Background()
+ testKey := "testKey"
+
+ for n := 0; n < b.N; n++ {
+ client.Get(ctx, testKey)
+ }
+}
+
+func BenchmarkClientSet(b *testing.B) {
+ tempDir := newTempDir(b)
+ dbFile := filepath.Join(tempDir, "my_db")
+
+ client, err := newClient(dbFile, time.Second)
+ require.NoError(b, err)
+
+ ctx := context.Background()
+ testKey := "testKey"
+ testValue := []byte("testValue")
+
+ for n := 0; n < b.N; n++ {
+ client.Set(ctx, testKey, testValue)
+ }
+}
+
+func BenchmarkClientDelete(b *testing.B) {
+ tempDir := newTempDir(b)
+ dbFile := filepath.Join(tempDir, "my_db")
+
+ client, err := newClient(dbFile, time.Second)
+ require.NoError(b, err)
+
+ ctx := context.Background()
+ testKey := "testKey"
+
+ for n := 0; n < b.N; n++ {
+ client.Delete(ctx, testKey)
+ }
+}
+
+func newTempDir(tb testing.TB) string {
+ tempDir, err := ioutil.TempDir("", "")
+ require.NoError(tb, err)
+ tb.Cleanup(func() { os.RemoveAll(tempDir) })
+ return tempDir
+}

extension/storage/filestorage/config.go

@@ -0,0 +1,29 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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
+//
+// https://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 filestorage
+
+import (
+ "time"
+
+ "go.opentelemetry.io/collector/config"
+)
+
+// Config defines configuration for http forwarder extension.
+type Config struct {
+ config.ExtensionSettings `mapstructure:",squash"`
+
+ Directory string `mapstructure:"directory,omitempty"`
+ Timeout time.Duration `mapstructure:"timeout,omitempty"`
+}