Resolve "Too many open files"

See merge request !1074

application-framework/registration/registration.go

@@ -2,6 +2,7 @@ package registration
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"code.fbi.h-da.de/danet/gosdn/api/go/gosdn/app"
@@ -15,6 +16,12 @@ func Register(ctx context.Context, gosdnAddress, name, token string) (string, er
 	if err != nil {
 		return "", err
 	}
+	defer func() {
+		if ferr := conn.Close(); ferr != nil {
+			fErrString := ferr.Error()
+			err = fmt.Errorf("InternalError=%w DeferError=%+s", err, fErrString)
+		}
+	}()
 
 	appService := app.NewAppServiceClient(conn)
 

controller/northbound/client/app.go

@@ -5,13 +5,19 @@ import (
 	"google.golang.org/grpc"
 )
 
+var appClientConnection *grpc.ClientConn
+
 // AppClient returns a client for the gRPC App service. It takes
 // the address of the gRPC endpoint and optional grpc.DialOption
 // as argument.
 func AppClient(addr string, opts ...grpc.DialOption) (apb.AppServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if appClientConnection == nil {
+		appClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return apb.NewAppServiceClient(conn), nil
+	return apb.NewAppServiceClient(appClientConnection), nil
 }

controller/northbound/client/configurationManagement.go

@@ -5,13 +5,20 @@ import (
 	"google.golang.org/grpc"
 )
 
+var configurationManagementClientConnection *grpc.ClientConn
+
 // ConfigurationManagementClient returns a client for the gRPC ConfigurationManagement service. It takes
 // the address of the gRPC endpoint and optional grpc.DialOption
 // as argument.
 func ConfigurationManagementClient(addr string, opts ...grpc.DialOption) (cpb.ConfigurationManagementServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if configurationManagementClientConnection == nil {
+		configurationManagementClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return cpb.NewConfigurationManagementServiceClient(conn), nil
+
+	return cpb.NewConfigurationManagementServiceClient(configurationManagementClientConnection), nil
 }

controller/northbound/client/networkElement.go

@@ -5,13 +5,20 @@ import (
 	"google.golang.org/grpc"
 )
 
+var networkElementServiceClientConnection *grpc.ClientConn
+
 // NetworkElementClient returns a client for the gRPC NetworkElement service. It takes
 // the address of the gRPC endpoint and optional grpc.DialOption
 // as argument.
 func NetworkElementClient(addr string, opts ...grpc.DialOption) (mnepb.NetworkElementServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if networkElementServiceClientConnection == nil {
+		networkElementServiceClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return mnepb.NewNetworkElementServiceClient(conn), nil
+
+	return mnepb.NewNetworkElementServiceClient(networkElementServiceClientConnection), nil
 }

controller/northbound/client/plugin.go

@@ -5,10 +5,16 @@ import (
 	"google.golang.org/grpc"
 )
 
+var pluginClientConnection *grpc.ClientConn
+
 func PluginClient(addr string, opts ...grpc.DialOption) (pipb.PluginInternalServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if pluginClientConnection == nil {
+		pluginClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return pipb.NewPluginInternalServiceClient(conn), nil
+	return pipb.NewPluginInternalServiceClient(pluginClientConnection), nil
 }

controller/northbound/client/pnd.go

@@ -5,13 +5,20 @@ import (
 	"google.golang.org/grpc"
 )
 
+var pndClientConnection *grpc.ClientConn
+
 // PndClient returns a client for the gRPC PND service. It takes
 // the address of the gRPC endpoint and optional grpc.DialOption
 // as argument.
 func PndClient(addr string, opts ...grpc.DialOption) (ppb.PndServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if pndClientConnection == nil {
+		pndClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return ppb.NewPndServiceClient(conn), nil
+
+	return ppb.NewPndServiceClient(pndClientConnection), nil
 }

controller/northbound/client/rbac.go

@@ -5,35 +5,51 @@ import (
 	"google.golang.org/grpc"
 )
 
+var authClientConnection *grpc.ClientConn
+var userClientConnection *grpc.ClientConn
+var roleClientConnection *grpc.ClientConn
+
 // AuthClient returns a client for the gRPC Auth service. It takes
 // the address of the gRPC endpoint and optional grpc.DialOption
 // as argument.
 func AuthClient(addr string, opts ...grpc.DialOption) (apb.AuthServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if authClientConnection == nil {
+		authClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return apb.NewAuthServiceClient(conn), nil
+	return apb.NewAuthServiceClient(authClientConnection), nil
 }
 
 // UserClient returns a client for the gRPC User service. It takes
 // the address of the gRPC endpoint and optional grpc.DialOption
 // as argument.
 func UserClient(addr string, opts ...grpc.DialOption) (apb.UserServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if userClientConnection == nil {
+		userClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return apb.NewUserServiceClient(conn), nil
+	return apb.NewUserServiceClient(userClientConnection), nil
 }
 
 // RoleClient returns a client for the gRPC Role service. It takes
 // the address of the gRPC endpoint and optional grpc.DialOption
 // as argument.
 func RoleClient(addr string, opts ...grpc.DialOption) (apb.RoleServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if roleClientConnection == nil {
+		roleClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return apb.NewRoleServiceClient(conn), nil
+	return apb.NewRoleServiceClient(roleClientConnection), nil
 }

controller/northbound/client/sbi.go

@@ -5,13 +5,20 @@ import (
 	"google.golang.org/grpc"
 )
 
+var sbiClientConnection *grpc.ClientConn
+
 // SbiClient returns a client for the gRPC SBI service. It takes
 // the address of the gRPC endpoint and optional grpc.DialOption
 // as argument.
 func SbiClient(addr string, opts ...grpc.DialOption) (spb.SbiServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if sbiClientConnection == nil {
+		sbiClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return spb.NewSbiServiceClient(conn), nil
+
+	return spb.NewSbiServiceClient(sbiClientConnection), nil
 }

controller/northbound/client/submanagement.go

@@ -5,13 +5,20 @@ import (
 	"google.golang.org/grpc"
 )
 
+var subManagementClientConnection *grpc.ClientConn
+
 // SubManagementClient returns a client for the gRPC SubscriptionManagement service. It takes
 // the address of the gRPC endpoint and optional grpc.DialOption
 // as argument.
 func SubManagementClient(addr string, opts ...grpc.DialOption) (subpb.SubscriptionManagementServiceClient, error) {
-	conn, err := grpc.NewClient(addr, opts...)
-	if err != nil {
-		return nil, err
+	var err error
+
+	if subManagementClientConnection == nil {
+		subManagementClientConnection, err = grpc.NewClient(addr, opts...)
+		if err != nil {
+			return nil, err
+		}
 	}
-	return subpb.NewSubscriptionManagementServiceClient(conn), nil
+
+	return subpb.NewSubscriptionManagementServiceClient(subManagementClientConnection), nil
 }

controller/nucleus/databaseNetworkElementStore.go

@@ -6,7 +6,6 @@ import (
 
 	"code.fbi.h-da.de/danet/gosdn/controller/customerrs"
 	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/networkelement"
-	"code.fbi.h-da.de/danet/gosdn/controller/nucleus/database"
 	"code.fbi.h-da.de/danet/gosdn/controller/store"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
@@ -122,15 +121,10 @@ func (s *DatabaseNetworkElementStore) Add(ctx context.Context, networkElementToA
 func (s *DatabaseNetworkElementStore) Update(ctx context.Context, networkElementToUpdate networkelement.NetworkElement) (err error) {
 	var updatedLoadedNetworkElement networkelement.LoadedNetworkElement
 
-	db, err := database.GetDatabaseConnection()
-	if err != nil {
-		return err
-	}
-
 	wc := writeconcern.Majority()
 	txnOptions := options.Transaction().SetWriteConcern(wc)
 	// Starts a session on the client
-	session, err := db.Client().StartSession()
+	session, err := s.collection.Database().Client().StartSession()
 	if err != nil {
 		return err
 	}

controller/nucleus/gnmi_transport.go

@@ -20,6 +20,8 @@ import (
 	tpb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/transport"
 )
 
+var gnmiClients = make(map[string]gpb.GNMIClient, 0)
+
 // Gnmi implements the Transport interface and provides an SBI with the
 // possibility to access a gNMI endpoint.
 type Gnmi struct {
@@ -64,9 +66,14 @@ func newGnmiTransport(opts *tpb.TransportOption, model shared.DeviceModel) (*Gnm
 		}
 	}
 
-	c, err := gnmi.Dial(gnmiConfig)
-	if err != nil {
-		return nil, err
+	var err error
+	c, ok := gnmiClients[opts.GetAddress()]
+	if !ok {
+		c, err = gnmi.Dial(gnmiConfig)
+		if err != nil {
+			return nil, err
+		}
+		gnmiClients[opts.GetAddress()] = c
 	}
 
 	log.WithFields(log.Fields{

controller/nucleus/plugin.go

@@ -16,6 +16,13 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 )
 
+type pluginConnection struct {
+	client *hcplugin.Client
+	model  shared.DeviceModel
+}
+
+var pluginClients = make(map[uuid.UUID]pluginConnection, 0)
+
 // Plugin is the controllers internal representation of a plugin.
 type Plugin struct {
 	UUID     uuid.UUID
@@ -66,6 +73,11 @@ func NewPlugin(id uuid.UUID, execPath string) (*Plugin, error) {
 		}
 	}
 
+	pluginClients[id] = pluginConnection{
+		client: client,
+		model:  model,
+	}
+
 	return &Plugin{
 		UUID:        id,
 		client:      client,
@@ -78,43 +90,52 @@ func NewPlugin(id uuid.UUID, execPath string) (*Plugin, error) {
 
 // NewPluginThroughReattachConfig creates a new Plugin through a reattach config.
 func NewPluginThroughReattachConfig(loadedPlugin plugin.LoadedPlugin) (plugin.Plugin, error) {
-	client := hcplugin.NewClient(&hcplugin.ClientConfig{
-		HandshakeConfig:  shared.Handshake,
-		Plugins:          shared.PluginMap,
-		Reattach:         &loadedPlugin.ReattachConfig,
-		AllowedProtocols: []hcplugin.Protocol{hcplugin.ProtocolGRPC},
-	})
-
-	// create a client that is within the AllowedProtocols. In this case this
-	// returns a gRPCClient. Allows to connect through gRPC.
-	gRPCClient, err := client.Client()
+	//client := hcplugin.NewClient(&hcplugin.ClientConfig{
+	//	HandshakeConfig:  shared.Handshake,
+	//	Plugins:          shared.PluginMap,
+	//	Reattach:         &loadedPlugin.ReattachConfig,
+	//	AllowedProtocols: []hcplugin.Protocol{hcplugin.ProtocolGRPC},
+	//})
+
+	//// create a client that is within the AllowedProtocols. In this case this
+	//// returns a gRPCClient. Allows to connect through gRPC.
+	//gRPCClient, err := client.Client()
+	//if err != nil {
+	//	return nil, err
+	//}
+
+	//// Request the plugin. This returns the gRPC client from the
+	//// DeviceModelPlugin. This can then be casted to the interface that we are
+	//// exposing through the plugin (in this case "DeviceModel").
+	//raw, err := gRPCClient.Dispense("deviceModel")
+	//if err != nil {
+	//	return nil, err
+	//}
+
+	//// cast the raw plugin to the DeviceModel interface. This allows to call
+	//// methods on the plugin as if it were a normal DeviceModel instance but
+	//// actually they are executed on the plugin sent through gRPC.
+	//model, ok := raw.(shared.DeviceModel)
+	//if !ok {
+	//	return nil, customerrs.InvalidTypeAssertionError{
+	//		Value: model,
+	//		Type:  (*shared.DeviceModel)(nil),
+	//	}
+	//}
+	pluginId, err := uuid.Parse(loadedPlugin.ID)
 	if err != nil {
 		return nil, err
 	}
 
-	// Request the plugin. This returns the gRPC client from the
-	// DeviceModelPlugin. This can then be casted to the interface that we are
-	// exposing through the plugin (in this case "DeviceModel").
-	raw, err := gRPCClient.Dispense("deviceModel")
-	if err != nil {
-		return nil, err
-	}
-
-	// cast the raw plugin to the DeviceModel interface. This allows to call
-	// methods on the plugin as if it were a normal DeviceModel instance but
-	// actually they are executed on the plugin sent through gRPC.
-	model, ok := raw.(shared.DeviceModel)
+	pc, ok := pluginClients[pluginId]
 	if !ok {
-		return nil, customerrs.InvalidTypeAssertionError{
-			Value: model,
-			Type:  (*shared.DeviceModel)(nil),
-		}
+		return nil, fmt.Errorf("plugin not found")
 	}
 
 	return &Plugin{
 		UUID:        uuid.MustParse(loadedPlugin.ID),
-		client:      client,
-		DeviceModel: model,
+		client:      pc.client,
+		DeviceModel: pc.model,
 		manifest:    &loadedPlugin.Manifest,
 		state:       plugin.INITIALIZED,
 	}, nil

controller/rbac/databaseUserStore.go

@@ -6,7 +6,6 @@ import (
 
 	"code.fbi.h-da.de/danet/gosdn/controller/customerrs"
 	"code.fbi.h-da.de/danet/gosdn/controller/interfaces/rbac"
-	"code.fbi.h-da.de/danet/gosdn/controller/nucleus/database"
 	"code.fbi.h-da.de/danet/gosdn/controller/store"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
@@ -134,15 +133,10 @@ func (s *DatabaseUserStore) GetAll(ctx context.Context) (loadedUsers []rbac.Load
 func (s *DatabaseUserStore) Update(ctx context.Context, userToUpdate rbac.User) (err error) {
 	var updatedLoadedUser rbac.LoadedUser
 
-	db, err := database.GetDatabaseConnection()
-	if err != nil {
-		return err
-	}
-
 	wc := writeconcern.Majority()
 	txnOptions := options.Transaction().SetWriteConcern(wc)
 	// Starts a session on the client
-	session, err := db.Client().StartSession()
+	session, err := s.collection.Database().Client().StartSession()
 	if err != nil {
 		return err
 	}

controller/topology/nodes/databaseNodeStore.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"code.fbi.h-da.de/danet/gosdn/controller/customerrs"
-	"code.fbi.h-da.de/danet/gosdn/controller/nucleus/database"
 	query "code.fbi.h-da.de/danet/gosdn/controller/store"
 
 	"github.com/google/uuid"
@@ -133,15 +132,10 @@ func (s *DatabaseNodeStore) Add(ctx context.Context, node Node) (err error) {
 func (s *DatabaseNodeStore) Update(ctx context.Context, node Node) (err error) {
 	var updatedLoadedNodes Node
 
-	db, err := database.GetDatabaseConnection()
-	if err != nil {
-		return err
-	}
-
 	wc := writeconcern.Majority()
 	txnOptions := options.Transaction().SetWriteConcern(wc)
 	// Starts a session on the client
-	session, err := db.Client().StartSession()
+	session, err := s.collection.Database().Client().StartSession()
 	if err != nil {
 		return err
 	}