Enhancement: Add the ability to save and restore the state of an adaptor

[jipperinbham]

Currently, if transporter fails, it is not able to start back where it left off during a copy/tail. In order to support this, each adaptor will want to save/persist the most recent document it has processed. We should be able to support multiple types of persistent stores using a simple interface like so:

type SessionStore interface {
    Set(path string, msg *message.Msg) error
    Get(path string) (string, int64, error)
}

The path would typically be a combination of the Transporter key and the node path. When retrieving the last known State, we will only return the last _id and the timestamp of the operation. It may be necessary in the future to support retuning the entire message.Msg.

Ideally, an implementing class will not constantly write the last operation but have the ability to "flush" on an interval which can be defined in the config.yaml.

The beginning of this can be seen in adaptor-state branch but it is incomplete. It currently introduces a sessionTicker where the Pipeline will call the Set func for each Node. As of right now, I have not added the ability to retrieve/get the state during startup/initialization.

Thoughts and feedback?

[jipperinbham]

With the changes being made in noid, it will no longer be viable to only store the id and timestamp. We'll need to store the entire Msg.Data now which will change the interface to

type SessionStore interface {
    Set(path string, msg *message.Msg) error
    Get(path string) (*message.Msg, error)
}

[jipperinbham]

Another part of this that has yet to be discussed is how to actually configure a SessionStore. My thought is to have something like the following in the config.yaml:

sessions:
  type: "filestore",
  interval: 10s,
  uri: "file:///tmp/transporter.state"

If the sessions section is missing, no SessionStore will be used.

[jipperinbham]

Made another change to the interface to use a struct instead of just the message.Msg. This will allow adaptors to store additional information while processing documents.

type SessionStore interface {
    Set(path string, state *MsgState) error
    Get(path string) (*MsgState, error)
}

type MsgState struct {
    Msg   *message.Msg
    Extra map[string]interface{}
}

[nstott]

do we need a map[string]interface{} there?
or can we assume we have a few specific states that we care about and use a type State int to store it?

[jipperinbham]

possibly, the purpose of the map is so adaptors could set things like Extra["progress"] = "copy" or Extra["progress"] = "tail" in the case of the MongoDB adaptor

that could just as easily be solved I guess but didn't want to limit things right away

[methuz]

This is the most important feature I need. I also want It to be distributed, Allow another node to restore the state if the current node is down by saving the state on a central server (or using etcd)

[tombray]

+1

[Alino]

+1

[jipperinbham]

@methuz I hope to get the adaptor state branch going again very soon but the initial implementation will not involve any distributed aspect to running transporter.

The initial goal is for transporter to save its running state at specified intervals to a SessionStore so if the transporter process crashed, it could be restarted and pick back up where it left off. In addition, the first version will likely only support 2 SessionStore implementations: File and Redis but more will be added over time.

[ambasta]

Hi,

Since the feature has already been merged into master, post-configuring the config.yaml, how does one go about using it in application.js?

[SherClockHolmes]

+1

[patmanh]

+1