Commit 1530ad32 authored by Carlos Natalino's avatar Carlos Natalino
Browse files

Addressing requests from review on the 10th of May.

parent fa7f2a66

my_deploy.sh

+2 −6
Original line number Diff line number Diff line
@@ -20,7 +20,7 @@ 
export TFS_REGISTRY_IMAGES="http://localhost:32000/tfs/"



# Set the list of components, separated by spaces, you want to build images for, and deploy.

export TFS_COMPONENTS="context device automation monitoring pathcomp service slice compute webui load_generator"

export TFS_COMPONENTS="context device automation monitoring pathcomp service slice compute webui load_generator l3_attackmitigator l3_centralizedattackdetector"



# Set the tag you want to use for your images.

export TFS_IMAGE_TAG="dev"
@@ -29,7 +29,7 @@ export TFS_IMAGE_TAG="dev" 
export TFS_K8S_NAMESPACE="tfs"



# Set additional manifest files to be applied after the deployment

export TFS_EXTRA_MANIFESTS="manifests/nginx_ingress_http.yaml manifests/servicemonitors.yaml manifests/cachingservice.yaml"

export TFS_EXTRA_MANIFESTS="manifests/nginx_ingress_http.yaml manifests/servicemonitors.yaml"



# Set the new Grafana admin password

export TFS_GRAFANA_PASSWORD="admin123+"
@@ -37,10 +37,6 @@ export TFS_GRAFANA_PASSWORD="admin123+" 
# Disable skip-build flag to rebuild the Docker images.

export TFS_SKIP_BUILD=""



# addition for the optical cybersecurity component

# export TFS_COMPONENTS="${TFS_COMPONENTS} dbscanserving opticalattackmitigator opticalattackdetector opticalattackmanager"

# export TFS_EXTRA_MANIFESTS="${TFS_EXTRA_MANIFESTS} manifests/cachingservice.yaml"





# ----- CockroachDB ------------------------------------------------------------

src/dbscanserving/Config.py

+0 −11
Original line number Diff line number Diff line
@@ -11,14 +11,3 @@ 
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.
 


import logging



# General settings

LOG_LEVEL = logging.DEBUG



# gRPC settings

GRPC_SERVICE_PORT = 10008



# Prometheus settings

METRICS_PORT = 9192

src/dbscanserving/client/DbscanServingClient.py

+9 −5
Original line number Diff line number Diff line
@@ -12,14 +12,16 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



import grpc, logging

import logging

from typing import Counter



import grpc



from common.Constants import ServiceNameEnum

from common.Settings import get_service_host, get_service_port_grpc

from common.tools.client.RetryDecorator import delay_exponential, retry

from common.proto.dbscanserving_pb2 import DetectionRequest, DetectionResponse

from common.proto.dbscanserving_pb2_grpc import DetectorStub

from common.Settings import get_service_host, get_service_port_grpc

from common.tools.client.RetryDecorator import delay_exponential, retry



LOGGER = logging.getLogger(__name__)

MAX_RETRIES = 15
@@ -33,8 +35,10 @@ RETRY_DECORATOR = retry( 


class DbscanServingClient:

    def __init__(self, host=None, port=None):

        if not host: host = get_service_host(ServiceNameEnum.DBSCANSERVING)

        if not port: port = get_service_port_grpc(ServiceNameEnum.DBSCANSERVING)

        if not host:

            host = get_service_host(ServiceNameEnum.DBSCANSERVING)

        if not port:

            port = get_service_port_grpc(ServiceNameEnum.DBSCANSERVING)



        self.endpoint = "{:s}:{:s}".format(str(host), str(port))

        LOGGER.debug("Creating channel to {:s}...".format(str(self.endpoint)))

src/dbscanserving/service/DbscanService.py

+10 −65
Original line number Diff line number Diff line
@@ -13,79 +13,24 @@ 
# limitations under the License.



import logging

from concurrent import futures



import grpc

from grpc_health.v1.health import OVERALL_HEALTH, HealthServicer

from grpc_health.v1.health_pb2 import HealthCheckResponse

from grpc_health.v1.health_pb2_grpc import add_HealthServicer_to_server



from common.Constants import (DEFAULT_GRPC_GRACE_PERIOD,

                              DEFAULT_GRPC_MAX_WORKERS)

from common.Constants import ServiceNameEnum

from common.proto.dbscanserving_pb2_grpc import add_DetectorServicer_to_server

from dbscanserving.Config import GRPC_SERVICE_PORT

from common.Settings import get_service_port_grpc

from common.tools.service.GenericGrpcService import GenericGrpcService



from dbscanserving.service.DbscanServiceServicerImpl import \

    DbscanServiceServicerImpl



BIND_ADDRESS = "0.0.0.0"

LOGGER = logging.getLogger(__name__)





class DbscanService:

    def __init__(

        self,

        address=BIND_ADDRESS,

        port=GRPC_SERVICE_PORT,

        grace_period=DEFAULT_GRPC_GRACE_PERIOD,

        max_workers=DEFAULT_GRPC_MAX_WORKERS,

    ):



        self.address = address

        self.port = port

        self.endpoint = None

        self.max_workers = max_workers

        self.grace_period = grace_period

        self.dbscan_servicer = None

        self.health_servicer = None

        self.pool = None

        self.server = None



    def start(self):

        self.endpoint = "{:s}:{:s}".format(str(self.address), str(self.port))

        LOGGER.debug(

            "Starting Service (tentative endpoint: {:s}, max_workers: {:s})...".format(

                str(self.endpoint), str(self.max_workers)

            )

        )



        self.pool = futures.ThreadPoolExecutor(max_workers=self.max_workers)

        self.server = grpc.server(self.pool)  # , interceptors=(tracer_interceptor,))

class DbscanService(GenericGrpcService):

    def __init__(self, cls_name: str = __name__):



        port = get_service_port_grpc(ServiceNameEnum.DBSCANSERVING)

        super().__init__(port, cls_name=cls_name)

        self.dbscan_servicer = DbscanServiceServicerImpl()

        add_DetectorServicer_to_server(self.dbscan_servicer, self.server)



        self.health_servicer = HealthServicer(

            experimental_non_blocking=True,

            experimental_thread_pool=futures.ThreadPoolExecutor(max_workers=1),

        )

        add_HealthServicer_to_server(self.health_servicer, self.server)



        port = self.server.add_insecure_port(self.endpoint)

        self.endpoint = "{:s}:{:s}".format(str(self.address), str(port))

        LOGGER.info("Listening on {:s}...".format(self.endpoint))

        self.server.start()

        self.health_servicer.set(

            OVERALL_HEALTH, HealthCheckResponse.SERVING

        )  # pylint: disable=maybe-no-member



        LOGGER.debug("Service started")



    def stop(self):

        LOGGER.debug(

            "Stopping service (grace period {:s} seconds)...".format(

                str(self.grace_period)

            )

        )

        self.health_servicer.enter_graceful_shutdown()

        self.server.stop(self.grace_period)

        LOGGER.debug("Service stopped")
 
    def install_servicers(self):

        add_DetectorServicer_to_server(self.dbscan_servicer, self.server)

src/dbscanserving/service/DbscanServiceServicerImpl.py

+3 −2
Original line number Diff line number Diff line
@@ -17,13 +17,14 @@ import logging 
import grpc

from sklearn.cluster import DBSCAN



from common.method_wrappers.Decorator import (MetricsPool,

                                              safe_and_metered_rpc_method)

from common.proto.dbscanserving_pb2 import DetectionRequest, DetectionResponse

from common.proto.dbscanserving_pb2_grpc import DetectorServicer

from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method



LOGGER = logging.getLogger(__name__)



METRICS_POOL = MetricsPool('DBSCANServing', 'RPC')

METRICS_POOL = MetricsPool("DBSCANServing", "RPC")





class DbscanServiceServicerImpl(DetectorServicer):