Adds unique context per grpc call in controlproxy

intelsdi-x · Jun 10, 2016 · c0dac05 · c0dac05
1 parent 096bd7d
commit c0dac05
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 13 deletions.
diff --git a/Godeps/Godeps.json b/Godeps/Godeps.json
diff --git a/grpc/controlproxy/controlproxy.go b/grpc/controlproxy/controlproxy.go
@@ -36,11 +36,15 @@ var (
 	MAX_CONNECTION_TIMEOUT = 10 * time.Second
 )
 
+func getContext() context.Context {
+	cd, _ := context.WithTimeout(context.Background(), MAX_CONNECTION_TIMEOUT)
+	return cd
+}
+
 // Implements managesMetrics interface provided by scheduler and
 // proxies those calls to the grpc client.
 type ControlProxy struct {
 	Client rpc.MetricManagerClient
-	ctx    context.Context
 }
 
 func New(addr string, port int) (ControlProxy, error) {
@@ -49,15 +53,14 @@ func New(addr string, port int) (ControlProxy, error) {
 		return ControlProxy{}, err
 	}
 	c := rpc.NewMetricManagerClient(conn)
-	cd, _ := context.WithTimeout(context.Background(), MAX_CONNECTION_TIMEOUT)
-	return ControlProxy{Client: c, ctx: cd}, nil
+	return ControlProxy{Client: c}, nil
 }
 
 func (c ControlProxy) ExpandWildcards(namespace core.Namespace) ([]core.Namespace, serror.SnapError) {
 	req := &rpc.ExpandWildcardsRequest{
 		Namespace: common.ToNamespace(namespace),
 	}
-	reply, err := c.Client.ExpandWildcards(c.ctx, req)
+	reply, err := c.Client.ExpandWildcards(getContext(), req)
 	if err != nil {
 		return nil, serror.New(err)
 	}
@@ -69,7 +72,7 @@ func (c ControlProxy) ExpandWildcards(namespace core.Namespace) ([]core.Namespac
 }
 func (c ControlProxy) PublishMetrics(contentType string, content []byte, pluginName string, pluginVersion int, config map[string]ctypes.ConfigValue, taskID string) []error {
 	req := GetPubProcReq(contentType, content, pluginName, pluginVersion, config, taskID)
-	reply, err := c.Client.PublishMetrics(c.ctx, req)
+	reply, err := c.Client.PublishMetrics(getContext(), req)
 	var errs []error
 	if err != nil {
 		errs = append(errs, err)
@@ -82,7 +85,7 @@ func (c ControlProxy) PublishMetrics(contentType string, content []byte, pluginN
 
 func (c ControlProxy) ProcessMetrics(contentType string, content []byte, pluginName string, pluginVersion int, config map[string]ctypes.ConfigValue, taskID string) (string, []byte, []error) {
 	req := GetPubProcReq(contentType, content, pluginName, pluginVersion, config, taskID)
-	reply, err := c.Client.ProcessMetrics(c.ctx, req)
+	reply, err := c.Client.ProcessMetrics(getContext(), req)
 	var errs []error
 	if err != nil {
 		errs = append(errs, err)
@@ -115,7 +118,7 @@ func (c ControlProxy) CollectMetrics(mts []core.Metric, deadline time.Time, task
 		TaskID:  taskID,
 		AllTags: allTags,
 	}
-	reply, err := c.Client.CollectMetrics(c.ctx, req)
+	reply, err := c.Client.CollectMetrics(getContext(), req)
 	var errs []error
 	if err != nil {
 		errs = append(errs, err)
@@ -136,7 +139,7 @@ func (c ControlProxy) GetPluginContentTypes(n string, t core.PluginType, v int)
 		PluginType: getPluginType(t),
 		Version:    int32(v),
 	}
-	reply, err := c.Client.GetPluginContentTypes(c.ctx, req)
+	reply, err := c.Client.GetPluginContentTypes(getContext(), req)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -151,7 +154,7 @@ func (c ControlProxy) ValidateDeps(mts []core.Metric, plugins []core.SubscribedP
 		Metrics: common.NewMetrics(mts),
 		Plugins: common.ToSubPluginsMsg(plugins),
 	}
-	reply, err := c.Client.ValidateDeps(c.ctx, req)
+	reply, err := c.Client.ValidateDeps(getContext(), req)
 	if err != nil {
 		return []serror.SnapError{serror.New(err)}
 	}
@@ -161,7 +164,7 @@ func (c ControlProxy) ValidateDeps(mts []core.Metric, plugins []core.SubscribedP
 
 func (c ControlProxy) SubscribeDeps(taskID string, mts []core.Metric, plugins []core.Plugin) []serror.SnapError {
 	req := depsRequest(taskID, mts, plugins)
-	reply, err := c.Client.SubscribeDeps(c.ctx, req)
+	reply, err := c.Client.SubscribeDeps(getContext(), req)
 	if err != nil {
 		return []serror.SnapError{serror.New(err)}
 	}
@@ -171,7 +174,7 @@ func (c ControlProxy) SubscribeDeps(taskID string, mts []core.Metric, plugins []
 
 func (c ControlProxy) UnsubscribeDeps(taskID string, mts []core.Metric, plugins []core.Plugin) []serror.SnapError {
 	req := depsRequest(taskID, mts, plugins)
-	reply, err := c.Client.UnsubscribeDeps(c.ctx, req)
+	reply, err := c.Client.UnsubscribeDeps(getContext(), req)
 	if err != nil {
 		return []serror.SnapError{serror.New(err)}
 	}
@@ -183,7 +186,7 @@ func (c ControlProxy) MatchQueryToNamespaces(namespace core.Namespace) ([]core.N
 	req := &rpc.ExpandWildcardsRequest{
 		Namespace: common.ToNamespace(namespace),
 	}
-	reply, err := c.Client.MatchQueryToNamespaces(c.ctx, req)
+	reply, err := c.Client.MatchQueryToNamespaces(getContext(), req)
 	if err != nil {
 		return nil, serror.New(err)
 	}