Add data provider

.github/workflows/ci.yml

@@ -49,6 +49,29 @@ jobs:
         env:
           TARGETPLATFORM: linux/amd64
 
+  data-provider-codegen-check:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.10.16
+
+      - name: Run code generation script
+        run: |
+          python ./apps/scripts/update_shared_data_provider_code.py
+
+      - name: Check for changes
+        run: |
+          if [[ $(git status --porcelain) ]]; then
+            echo "Generated code is out of sync. Please run the script and commit the changes."
+            exit 1
+          fi
+
   test-evm:
     runs-on: ubuntu-latest
 

apps/README.md

@@ -17,3 +17,13 @@ The Stork Network receives signed data feeds from publishers and aggregates them
 The easiest way to become a Stork Publisher is to run the Stork Publisher Agent docker container on your infrastructure and send price updates to the Agent through a local websocket. The Stork Publisher Agent will sign your price updates with your private key and send them to the Stork Network.
 
 See [Stork Publisher Agent Docs](docs/publisher_agent.md).
+
+## Data Provider
+
+To publish data into the Stork Network, a Publisher first needs to fetch that data from some data source.
+
+The Stork Data Provider is an app that lets users configure a list of data feeds from various sources which they would like to output. These data streams are output in a format which can be easily received by the Publisher Agent, meaning a user can run the Data Provider alongside the Publisher Agent so that they can source the data, sign it and send it to the Stork Network without writing any code.
+
+It is also an open source framework where users can easily contribute to a collection of data integrations.
+
+See [Stork Data Provider Docs](docs/data_provider.md).

apps/docs/data_provider.md

@@ -0,0 +1,52 @@
+# Data Provider
+The Stork Data Provider is a framework to pull arbitrary numeric data across many sources. It can be used on its own, or run alongside the Stork Publisher Agent to sign the data and send it to the Stork Network.
+
+## Adding a New Data Source
+If you want to report data from a data source which does not already have an [integration](../lib/data_provider/sources), you can add your own.
+
+To add a new source:
+1. Add a [package](../lib/data_provider/sources/random) in the [sources directory](../lib/data_provider/sources) with your data source's name
+1. Run `python3 ./apps/scripts/update_shared_data_provider_code/py` to generate some framework code so that the framework is aware of your new source.
+1. Add a [data_source.go](../lib/data_provider/sources/random/data_source.go) and implement a DataSource object conforming to the [DataSource interface](../lib/data_provider/types/model.go). This object will contain most of your source-specific logic, but it can leverage tools like the [scheduler](../lib/data_provider/sources/scheduler.go) or [ethereum_utils](../lib/data_provider/sources/ethereum_utils.go) as needed.
+1. Add a [data_source_test.go](../lib/data_provider/sources/random/data_source_test.go) to unit test your data source.
+1. Add a [config.go](../lib/data_provider/sources/random/config.go) which defines a configuration object corresponding to a single data feed in your source
+   1. This config object must include a `DataSource` field.
+1. Add a [JSON Schema](https://json-schema.org/) [config](../lib/data_provider/configs/resources/source_config_schemas/random.json) in the configs package defining the structure of the configuration object in [config.go](../lib/data_provider/sources/random/config.go)
+1. Add a [config test](../lib/data_provider/configs/source_config_tests/random_test.go) to the configs package which tests that a valid Data Provider config json using your source:
+   1. Passes schema validations
+   1. Can be deserialized into your configuration object correctly
+   1. Can be used to extract your DataSourceId using `GetSourceSpecificConfig`
+1. Add an [init.go](../lib/data_provider/sources/random/init.go) to your package. This file can be almost identical for every source. This file is responsible for:
+   1. Defining the DataSourceId variable for this source (which must be the same as the package name)
+   1. Defining and registering a DataSourceFactory (which will just call to your DataSource constructor)
+   1. Asserting the source's DataSource and DataSourceFactory satisfy our interfaces
+   1. Defining a function to deserialize the source's config object
+1. Submit a Pull Request so other developers can use your new data source!
+
+## Configuration
+The Data Provider can report many feeds, each sourced from any of the data sources implemented in [sources](../lib/data_provider/sources).
+
+You can configure the Data Provider by passing it a [config json file](../../sample.data-provider.config.json) which can be deserialized into a [DataProviderConfig](../lib/data_provider/types/model.go) object. 
+
+The `sources` tag is a list of configurations for different feeds, where each feed has a unique `id` and a `config` which can be deserialized into the appropriate [source config](../lib/data_provider/sources/random/config.go). 
+
+## Running Local Code
+You can test the Data Provider locally by running:
+```
+go run apps/cmd/data_provider/main.go start -c ./sample.data-provider.config.json --verbose
+```
+You will most likely want to replace the `./sample.data-provider.config.json` with a more useful config json. Also make sure any required environment variables like API keys are set in your local environment.
+
+Running in `--verbose` mode with no output address set will just log every price update. If you want to actually send updates somewhere (like the websocket server of your local Publisher Agent), you can pass an output address flag:
+```
+go run apps/cmd/data_provider/main.go start -c ./sample.data-provider.config.json -o ws://localhost:5216/
+```
+
+## Running Published Docker Image
+If all the data sources you want to use are already merged into Stork's repo, you can just pull the latest published Data Provider docker image and supply your own config: 
+```
+docker run --platform linux/arm64 --pull always --restart always --name data-provider -v ./sample.data-provider.config.json:/etc/config.json -d --log-opt max-size=1g storknetwork/data-provider:v1.0.4 start -c /etc/config.json -o ws://localhost:5216/
+```
+
+
+

apps/lib/data_provider/command.go

@@ -18,18 +18,18 @@ var DataProviderCmd = &cobra.Command{
 
 // required
 const ConfigFilePathFlag = "config-file-path"
-const WebsocketUrl = "ws-url"
+const OutputAddressFlag = "output-address"
 
 func init() {
 	DataProviderCmd.Flags().StringP(ConfigFilePathFlag, "c", "", "the path of your config json file")
-	DataProviderCmd.Flags().StringP(WebsocketUrl, "w", "", "the websocket url to write updates to")
+	DataProviderCmd.Flags().StringP(OutputAddressFlag, "o", "", "a string representing an output address (e.g. ws://localhost:5216/)")
 
 	DataProviderCmd.MarkFlagRequired(ConfigFilePathFlag)
 }
 
 func runDataProvider(cmd *cobra.Command, args []string) error {
 	configFilePath, _ := cmd.Flags().GetString(ConfigFilePathFlag)
-	wsUrl, _ := cmd.Flags().GetString(WebsocketUrl)
+	outputAddress, _ := cmd.Flags().GetString(OutputAddressFlag)
 
 	mainLogger := utils.MainLogger()
 
@@ -39,12 +39,12 @@ func runDataProvider(cmd *cobra.Command, args []string) error {
 
 	mainLogger.Info().Msg("Starting data provider")
 
-	config, err := loadConfig(configFilePath)
+	config, err := LoadConfig(configFilePath)
 	if err != nil {
 		return fmt.Errorf("error loading config: %v", err)
 	}
 
-	runner := NewDataProviderRunner(*config, wsUrl)
+	runner := NewDataProviderRunner(*config, outputAddress)
 	runner.Run()
 
 	return nil

apps/lib/data_provider/config.go

@@ -1,97 +1,18 @@
 package data_provider
 
 import (
-	"embed"
-	"encoding/json"
 	"fmt"
 	"os"
 
-	"github.com/Stork-Oracle/stork-external/apps/lib/data_provider/sources"
+	"github.com/Stork-Oracle/stork-external/apps/lib/data_provider/configs"
 	"github.com/Stork-Oracle/stork-external/apps/lib/data_provider/types"
-	"github.com/Stork-Oracle/stork-external/apps/lib/data_provider/utils"
-	"github.com/xeipuuv/gojsonschema"
 )
 
-//go:embed resources
-var resourcesFS embed.FS
-
-func loadConfig(configPath string) (*types.DataProviderConfig, error) {
+func LoadConfig(configPath string) (*types.DataProviderConfig, error) {
 	configBytes, err := os.ReadFile(configPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to read config file: %v", err)
 	}
 
-	err = validateConfig(configBytes)
-	if err != nil {
-		return nil, fmt.Errorf("config file is invalid: %v", err)
-	}
-
-	var config types.DataProviderConfig
-	if err := json.Unmarshal(configBytes, &config); err != nil {
-		return nil, fmt.Errorf("failed to unmarshal config file: %v", err)
-	}
-	return &config, nil
-}
-
-func validateSourceConfigs(sourceConfigsObj interface{}) error {
-	sourceConfigs, ok := sourceConfigsObj.([]interface{})
-	if !ok {
-		return fmt.Errorf("invalid source configs type: %T", sourceConfigsObj)
-	}
-
-	for _, sourceConfig := range sourceConfigs {
-		sourceConfigMap, ok := sourceConfig.(map[string]interface{})
-		if !ok {
-			return fmt.Errorf("invalid source config type: %v", sourceConfig)
-		}
-		dataSourceId := sourceConfigMap["dataSource"].(string)
-		factory, err := sources.GetDataSourceFactory(types.DataSourceId(dataSourceId))
-		if err != nil {
-			return err
-		}
-		schema, err := factory.GetSchema()
-		if err != nil {
-			return err
-		}
-
-		sourceSpecificConfig := sourceConfigMap["config"]
-		configLoader := gojsonschema.NewGoLoader(sourceSpecificConfig)
-		result, err := schema.Validate(configLoader)
-		if err != nil {
-			return fmt.Errorf("error validating config: %v", err)
-		}
-		if !result.Valid() {
-			return fmt.Errorf("config is invalid: %v", result.Errors())
-		}
-	}
-
-	return nil
-}
-
-func validateConfig(configBytes []byte) error {
-	var dataProviderConfig map[string]interface{}
-	if err := json.Unmarshal(configBytes, &dataProviderConfig); err != nil {
-		return fmt.Errorf("failed to parse config JSON: %v", err)
-	}
-
-	// validate top level of config
-	schema, err := utils.LoadSchema("resources/config_schema.json", resourcesFS)
-	if err != nil {
-		return fmt.Errorf("error loading schema: %v", err)
-	}
-
-	configLoader := gojsonschema.NewGoLoader(dataProviderConfig)
-	result, err := schema.Validate(configLoader)
-	if err != nil {
-		return fmt.Errorf("error validating config: %v", err)
-	}
-	if !result.Valid() {
-		return fmt.Errorf("config is invalid: %v", result.Errors())
-	}
-
-	err = validateSourceConfigs(dataProviderConfig["sources"])
-	if err != nil {
-		return fmt.Errorf("error validating source configs: %v", err)
-	}
-	return nil
+	return configs.LoadConfigFromBytes(configBytes)
 }

apps/lib/data_provider/configs/config.go

@@ -0,0 +1,90 @@
+package configs
+
+import (
+	"embed"
+	"encoding/json"
+	"fmt"
+	"path/filepath"
+
+	"github.com/Stork-Oracle/stork-external/apps/lib/data_provider/types"
+	"github.com/xeipuuv/gojsonschema"
+)
+
+//go:embed resources
+var resourcesFS embed.FS
+
+const configSchemaPath = "resources/data_provider_config.schema.json"
+
+// exposed for testing
+func LoadConfigFromBytes(configBytes []byte) (*types.DataProviderConfig, error) {
+	schema, err := loadSchema(resourcesFS)
+	if err != nil {
+		return nil, fmt.Errorf("error loading schema: %v", err)
+	}
+
+	err = validateConfig(configBytes, schema)
+	if err != nil {
+		return nil, fmt.Errorf("config file is invalid: %v", err)
+	}
+
+	var config types.DataProviderConfig
+	if err := json.Unmarshal(configBytes, &config); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal config file: %v", err)
+	}
+	return &config, nil
+}
+
+func loadSchema(resourcesFS embed.FS) (*gojsonschema.Schema, error) {
+	schemaContent, err := resourcesFS.ReadFile(configSchemaPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read schema file for %s: %v", configSchemaPath, err)
+	}
+
+	loader := gojsonschema.NewSchemaLoader()
+
+	// add all source schema configs to schema loader
+	sourceSchemaDir := "resources/source_config_schemas"
+	sourceSchemaFiles, err := resourcesFS.ReadDir(sourceSchemaDir)
+	if err != nil {
+		return nil, err
+	}
+	for _, sourceSchemaFile := range sourceSchemaFiles {
+		sourceSchemaPath := filepath.Join(sourceSchemaDir, sourceSchemaFile.Name())
+		schemaBytes, err := resourcesFS.ReadFile(sourceSchemaPath)
+		if err != nil {
+			return nil, err
+		}
+		schemaFileLoader := gojsonschema.NewBytesLoader(schemaBytes)
+		err = loader.AddSchema(sourceSchemaPath, schemaFileLoader)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	topLevelSchemaLoader := gojsonschema.NewStringLoader(string(schemaContent))
+
+	schema, err := loader.Compile(topLevelSchemaLoader)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse schema for %s: %v", configSchemaPath, err)
+	}
+
+	return schema, nil
+}
+
+func validateConfig(configBytes []byte, schema *gojsonschema.Schema) error {
+	var dataProviderConfig map[string]interface{}
+	if err := json.Unmarshal(configBytes, &dataProviderConfig); err != nil {
+		return fmt.Errorf("failed to parse config JSON: %v", err)
+	}
+
+	configLoader := gojsonschema.NewGoLoader(dataProviderConfig)
+	result, err := schema.Validate(configLoader)
+	if err != nil {
+		return fmt.Errorf("error validating config: %v", err)
+	}
+	if !result.Valid() {
+		return fmt.Errorf("config is invalid: %v", result.Errors())
+	}
+
+	return nil
+}