func()

in server/internal/server/api.go [63:193]
113 lines of code
31 McCabe index (conditional complexity)

func (s *Server) Init(options Options) {
	var err error
	var cred azcore.TokenCredential

	logger := log.New(log.NewTextHandler(os.Stdout, nil).WithAttrs(logattrs.GetAttrs()))
	if options.JsonLog {
		logger = log.New(log.NewJSONHandler(os.Stdout, nil).WithAttrs(logattrs.GetAttrs()))
	}

	log.SetDefault(logger)
	if options.EnableAzureSDKCalls {
		armClientOptions := serviceHubPolicy.GetDefaultArmClientOptions(logger)
		// Use MSI in Standalone E2E env for credential
		if options.IdentityResourceID != "" {
			resourceID := azidentity.ResourceID(options.IdentityResourceID)
			opts := azidentity.ManagedIdentityCredentialOptions{ID: resourceID}
			cred, err = azidentity.NewManagedIdentityCredential(&opts)
		} else {
			cred, err = azidentity.NewDefaultAzureCredential(nil)
		}
		if err != nil {
			log.Error(err.Error())
			os.Exit(1)
		}
		resourcesClientFactory, err := armresources.NewClientFactory(options.SubscriptionID, cred, armClientOptions)
		if err != nil {
			log.Error(err.Error())
			os.Exit(1)
		}

		s.ResourceGroupClient = resourcesClientFactory.NewResourceGroupsClient()
		s.AccountsClient, err = armstorage.NewAccountsClient(options.SubscriptionID, cred, armClientOptions)
		if err != nil {
			log.Error(err.Error())
			os.Exit(1)
		}
	}

	if options.RemoteAddr != "" {
		s.client, err = client.NewClient(options.RemoteAddr, interceptor.GetClientInterceptorLogOptions(logger, logattrs.GetAttrs()))
		// logging the error for transparency, retry interceptor will handle it
		if err != nil {
			log.Error("did not connect: " + err.Error())
		}
	}

	if options.ServiceBusHostName != "" {
		s.serviceBusClient, err = servicebus.CreateServiceBusClient(context.Background(), options.ServiceBusHostName, nil, nil)
		if err != nil {
			logger.Error("Something went wrong creating the service bus client: " + err.Error())
			os.Exit(1)
		}
	}

	if options.ServiceBusQueueName != "" {
		s.serviceBusSender, err = s.serviceBusClient.NewServiceBusSender(context.Background(), options.ServiceBusQueueName, nil)
		if err != nil {
			logger.Error("Something went wrong creating the service bus sender: " + err.Error())
			os.Exit(1)
		}
	}

	if options.OperationContainerAddr != "" {
		s.operationContainerClient, err = ocClient.NewClient(options.OperationContainerAddr, interceptor.GetClientInterceptorLogOptions(logger, logattrs.GetAttrs()))
		if err != nil {
			logger.Error("Failed to initialize operationContainerClient at " + options.OperationContainerAddr + "with err: " + err.Error())
			os.Exit(1)
		}
	}

	dbStarted := false
	if options.DatabaseConnectionString != "" {
		s.dbClient, err = database.NewDbClientWithConnectionString(context.Background(), options.DatabaseConnectionString)
		if err != nil {
			logger.Error("Error creating connection pool: " + err.Error())
			os.Exit(1)
		}
		dbStarted = true
	} else if options.DatabaseServerUrl != "" && options.DatabaseName != "" {
		s.dbClient, err = database.NewDbClient(context.Background(), options.DatabaseServerUrl, options.DatabasePort, options.DatabaseName)
		if err != nil {
			logger.Error("Error creating connection pool: " + err.Error())
			os.Exit(1)
		}
		dbStarted = true
	}

	if dbStarted {

		if options.EntityTableName == "" {
			logger.Error("No OperationTableName set.")
			os.Exit(1)
		}

		if err := sanitizeTableName(options.EntityTableName); err != nil {
			logger.Error("Table name is not valid: " + err.Error())
			os.Exit(1)
		}

		s.entityTableName = options.EntityTableName
		//TODO(mheberling): Move this common functionality to a public repo which can also house other util functions we use.
		// Check if table exists
		entityListCheckQuery := "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = @p1"
		rows, err := database.QueryDb(context.Background(), s.dbClient, entityListCheckQuery, options.EntityTableName)
		if err != nil {
			logger.Error("Error querying for entity table metadata: " + err.Error())
			os.Exit(1)
		}
		defer rows.Close()

		var TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, TABLE_TYPE string
		for rows.Next() {
			err = rows.Scan(&TABLE_CATALOG, &TABLE_SCHEMA, &TABLE_NAME, &TABLE_TYPE)
			if err != nil {
				logger.Error("Error getting the operationStatus of the current Entity: " + err.Error())
				os.Exit(1)
			}
		}

		if TABLE_CATALOG == "" && TABLE_SCHEMA == "" && TABLE_NAME == "" && TABLE_TYPE == "" {
			logger.Info(fmt.Sprintf("The table %s doesn't exist!", s.entityTableName))
			entityListCreateTableQuery := fmt.Sprintf("CREATE TABLE %s (entity_type VARCHAR(255), entity_id VARCHAR(255), last_operation_id VARCHAR(255), operation_name VARCHAR (255), operation_status VARCHAR(255))", s.entityTableName)
			_, err = database.QueryDb(context.Background(), s.dbClient, entityListCreateTableQuery)
			if err != nil {
				logger.Error("Error creating the entity table: " + err.Error())
				os.Exit(1)
			}
			logger.Info(fmt.Sprintf("The table %s has been created.", options.EntityTableName))
		}
	}
}