Remove unnecessary Session, one per ConfigSource anyway

This PR makes the ConfigSource equivalent with the Session, and
relies on the fact that when created ConfigSource does what
NewSession was doing.

Signed-off-by: Bogdan Drutu <[email protected]>

config/experimental/configsource/component.go

@@ -33,25 +33,18 @@ var ErrValueUpdated = errors.New("configuration must retrieve the updated value"
 
 // ConfigSource is the interface to be implemented by objects used by the collector
 // to retrieve external configuration information.
+//
+// ConfigSource object will be used to retrieve full configuration or data to be
+// injected into a configuration.
+//
+// The ConfigSource object should use its creation according to the source needs:
+// lock resources, open connections, etc. An implementation, for instance,
+// can use the creation time to prevent torn configurations, by acquiring a lock
+// (or some other mechanism) that prevents concurrent changes to the configuration
+// during time that data is being retrieved from the source.
+//
+// The code managing the ConfigSource instance must guarantee that the object is not used concurrently.
 type ConfigSource interface {
- // NewSession must create a Session object that will be used to retrieve data to
- // be injected into a configuration.
- //
- // The Session object should use its creation according to their ConfigSource needs:
- // lock resources, open connections, etc. An implementation, for instance,
- // can use the creation of the Session object to prevent torn configurations,
- // by acquiring a lock (or some other mechanism) that prevents concurrent changes to the
- // configuration during time that data is being retrieved from the source.
- //
- // The code managing the returned Session object must guarantee that the object is not used
- // concurrently and that a single ConfigSource only have one Session open at any time.
- NewSession(ctx context.Context) (Session, error)
-}
-
-// Session is the interface used to retrieve configuration data from a ConfigSource. A Session
-// object is created from a ConfigSource. The code using Session objects must guarantee that
-// methods of a single instance are not called concurrently.
-type Session interface {
  // Retrieve goes to the configuration source and retrieves the selected data which
  // contains the value to be injected in the configuration and the corresponding watcher that
  // will be used to monitor for updates of the retrieved value. The retrieved value is selected

config/internal/configsource/manager.go

@@ -157,9 +157,6 @@ type Manager struct {
  // configSources is map from ConfigSource names (as defined in the configuration)
  // and the respective instances.
  configSources map[string]configsource.ConfigSource
- // sessions track all the Session objects used to retrieve values to be injected
- // into the configuration.
- sessions map[string]configsource.Session
  // watchers keeps track of all WatchForUpdate functions for retrieved values.
  watchers []configsource.Watchable
  // watchersWG is used to ensure that Close waits for all WatchForUpdate calls
@@ -180,8 +177,6 @@ func NewManager(_ *configparser.Parser) (*Manager, error) {
  // TODO: Config sources should be extracted for the config itself, need Factories for that.
 
  return &Manager{
- // TODO: Temporarily tests should set their config sources per their needs.
- sessions: make(map[string]configsource.Session),
  watchingCh: make(chan struct{}),
  closeCh: make(chan struct{}),
  }, nil
@@ -196,14 +191,14 @@ func (m *Manager) Resolve(ctx context.Context, parser *configparser.Parser) (*co
  for _, k := range allKeys {
  value, err := m.expandStringValues(ctx, parser.Get(k))
  if err != nil {
- // Call RetrieveEnd for all sessions used so far but don't record any errors.
- _ = m.retrieveEndAllSessions(ctx)
+ // Call RetrieveEnd for all sources used so far but don't record any errors.
+ _ = m.retrieveEnd(ctx)
  return nil, err
  }
  res.Set(k, value)
  }
 
- if errs := m.retrieveEndAllSessions(ctx); len(errs) > 0 {
+ if errs := m.retrieveEnd(ctx); len(errs) > 0 {
  return nil, consumererror.Combine(errs)
  }
 
@@ -268,8 +263,8 @@ func (m *Manager) WaitForWatcher() {
 // in the configuration. It should be called
 func (m *Manager) Close(ctx context.Context) error {
  var errs []error
- for _, session := range m.sessions {
- if err := session.Close(ctx); err != nil {
+ for _, source := range m.configSources {
+ if err := source.Close(ctx); err != nil {
  errs = append(errs, err)
  }
  }
@@ -280,10 +275,10 @@ func (m *Manager) Close(ctx context.Context) error {
  return consumererror.Combine(errs)
 }
 
-func (m *Manager) retrieveEndAllSessions(ctx context.Context) []error {
+func (m *Manager) retrieveEnd(ctx context.Context) []error {
  var errs []error
- for _, session := range m.sessions {
- if err := session.RetrieveEnd(ctx); err != nil {
+ for _, source := range m.configSources {
+ if err := source.RetrieveEnd(ctx); err != nil {
  errs = append(errs, err)
  }
  }
@@ -337,16 +332,7 @@ func (m *Manager) expandConfigSource(ctx context.Context, cfgSrc configsource.Co
  return nil, err
  }
 
- session, ok := m.sessions[cfgSrcName]
- if !ok {
- session, err = cfgSrc.NewSession(ctx)
- if err != nil {
- return nil, fmt.Errorf("failed to create session for config source %q: %w", cfgSrcName, err)
- }
- m.sessions[cfgSrcName] = session
- }
-
- retrieved, err := session.Retrieve(ctx, selector, params)
+ retrieved, err := cfgSrc.Retrieve(ctx, selector, params)
  if err != nil {
  return nil, fmt.Errorf("config source %q failed to retrieve value: %w", cfgSrcName, err)
  }

config/internal/configsource/manager_test.go

@@ -91,15 +91,6 @@ func TestConfigSourceManager_ResolveErrors(t *testing.T) {
  "tstcfgsrc": &testConfigSource{},
  },
  },
- {
- name: "error_on_new_session",
- config: map[string]interface{}{
- "cfgsrc": "$tstcfgsrc:selector",
- },
- configSourceMap: map[string]configsource.ConfigSource{
- "tstcfgsrc": &testConfigSource{ErrOnNewSession: testErr},
- },
- },
  {
  name: "error_on_retrieve",
  config: map[string]interface{}{
@@ -563,7 +554,6 @@ func Test_parseCfgSrc(t *testing.T) {
 type testConfigSource struct {
  ValueMap map[string]valueEntry
 
- ErrOnNewSession error
  ErrOnRetrieve error
  ErrOnRetrieveEnd error
  ErrOnClose error
@@ -577,14 +567,6 @@ type valueEntry struct {
 }
 
 var _ configsource.ConfigSource = (*testConfigSource)(nil)
-var _ configsource.Session = (*testConfigSource)(nil)
-
-func (t *testConfigSource) NewSession(context.Context) (configsource.Session, error) {
- if t.ErrOnNewSession != nil {
- return nil, t.ErrOnNewSession
- }
- return t, nil
-}
 
 func (t *testConfigSource) Retrieve(ctx context.Context, selector string, params interface{}) (configsource.Retrieved, error) {
  if t.OnRetrieve != nil {