@@ -3,16 +3,24 @@ package rpc
33import (
44 "encoding/json"
55 "fmt"
6+ "net/url"
7+ "path"
68 "reflect"
79 "sync"
810
11+ "github.com/gojek/turing/engines/experiment/config"
912 "github.com/gojek/turing/engines/experiment/manager"
13+ "github.com/gojek/turing/engines/experiment/pkg/utils"
1014 "github.com/gojek/turing/engines/experiment/plugin/rpc/shared"
1115 "github.com/gojek/turing/engines/experiment/runner"
1216 "github.com/hashicorp/go-plugin"
17+ "github.com/mitchellh/hashstructure/v2"
1318 "go.uber.org/zap"
1419)
1520
21+ var factoriesmu sync.Mutex
22+ var factories = make (map [string ]* EngineFactory )
23+
1624// EngineFactory implements experiment.EngineFactory and creates experiment manager/runner
1725// backed by net/rpc plugin implementations
1826type EngineFactory struct {
@@ -74,7 +82,44 @@ func (f *EngineFactory) GetExperimentRunner() (runner.ExperimentRunner, error) {
7482 return f .runner , nil
7583}
7684
77- func NewFactory (pluginBinary string , engineCfg json.RawMessage , logger * zap.SugaredLogger ) (* EngineFactory , error ) {
85+ func NewFactory (name string , cfg config.EngineConfig , logger * zap.SugaredLogger ) (* EngineFactory , error ) {
86+ factoriesmu .Lock ()
87+ defer factoriesmu .Unlock ()
88+
89+ // get a hash of the engine's configuration and use it as a configuration's fingerprint
90+ cfgHash , err := hashstructure .Hash (cfg , hashstructure .FormatV2 , nil )
91+ if err != nil {
92+ return nil , err
93+ }
94+ factoryKey := fmt .Sprintf ("%s-%d" , name , cfgHash )
95+
96+ if engineFactory , ok := factories [factoryKey ]; ok {
97+ return engineFactory , nil
98+ }
99+
100+ engineCfg , err := cfg .RawEngineConfig ()
101+ if err != nil {
102+ return nil , err
103+ }
104+ if cfg .PluginBinary != "" {
105+ factories [factoryKey ], err = NewFactoryFromBinary (cfg .PluginBinary , engineCfg , logger )
106+ } else if cfg .PluginURL != "" {
107+ factories [factoryKey ], err = NewFactoryFromURL (cfg .PluginURL , engineCfg , logger )
108+ } else {
109+ err = fmt .Errorf ("either `plugin_url` or `plugin_binary` must be specified" )
110+ }
111+
112+ if err != nil {
113+ return nil , err
114+ }
115+ return factories [factoryKey ], nil
116+ }
117+
118+ func NewFactoryFromBinary (
119+ pluginBinary string ,
120+ engineCfg json.RawMessage ,
121+ logger * zap.SugaredLogger ,
122+ ) (* EngineFactory , error ) {
78123 rpcClient , err := Connect (pluginBinary , logger .Desugar ())
79124 if err != nil {
80125 return nil , err
@@ -85,3 +130,22 @@ func NewFactory(pluginBinary string, engineCfg json.RawMessage, logger *zap.Suga
85130 EngineConfig : engineCfg ,
86131 }, nil
87132}
133+
134+ func NewFactoryFromURL (
135+ pluginURL string ,
136+ engineCfg json.RawMessage ,
137+ logger * zap.SugaredLogger ,
138+ ) (* EngineFactory , error ) {
139+ downloadURL , err := url .Parse (pluginURL )
140+ if err != nil {
141+ return nil , fmt .Errorf ("failed to parse plugin URL: %v" , err )
142+ }
143+
144+ filename := fmt .Sprintf ("./%s" , path .Base (downloadURL .Path ))
145+ err = utils .DownloadFile (downloadURL , filename , 0744 )
146+ if err != nil {
147+ return nil , fmt .Errorf (
148+ "failed to download plugin's binary from remote url: url=%s, %v" , pluginURL , err )
149+ }
150+ return NewFactoryFromBinary (filename , engineCfg , logger )
151+ }
0 commit comments