Configurar el almacenamiento de mensajes con Milvus Operator

Milvus utiliza RocksMQ, Pulsar o Kafka para gestionar registros de cambios recientes, emitir registros de flujo y proporcionar suscripciones a registros. Este tema presenta cómo configurar las dependencias de almacenamiento de mensajes cuando instala Milvus con Milvus Operator. Para más detalles, consulte Configurar el almacenamiento de mensajes con Milvus Operator en el repositorio de Milvus Operator.

Este tema asume que usted ha desplegado Milvus Operator.

Necesita especificar un archivo de configuración para utilizar Milvus Operator para iniciar un cluster Milvus.

kubectl apply -f https://raw.githubusercontent.com/zilliztech/milvus-operator/main/config/samples/milvus_cluster_default.yaml

Sólo necesita editar la plantilla de código en milvus_cluster_default.yaml para configurar las dependencias de terceros. Las siguientes secciones presentan cómo configurar el almacenamiento de objetos, etcd y Pulsar respectivamente.

Antes de empezar

La siguiente tabla muestra si RocksMQ, NATS, Pulsar y Kafka son compatibles con Milvus en modo autónomo y en clúster.

También hay otras limitaciones para especificar el almacenamiento de mensajes:

• Sólo se admite un almacenamiento de mensajes para una instancia de Milvus. Sin embargo, todavía tenemos compatibilidad con múltiples almacenamientos de mensajes establecidos para una instancia. La prioridad es la siguiente
  • modo autónomo: RocksMQ (por defecto) > Pulsar > Kafka
  • modo clúster: Pulsar (por defecto) > Kafka
  • Los Nats introducidos en 2.3 no participan en estas reglas de prioridad por compatibilidad con versiones anteriores.
• El almacenamiento de mensajes no puede cambiarse mientras el sistema Milvus está en funcionamiento.
• Sólo se admite la versión 2.x o 3.x de Kafka.

Configurar RocksMQ

RocksMQ es el almacenamiento de mensajes por defecto en Milvus standalone.

Ejemplo

El siguiente ejemplo configura un servicio RocksMQ.

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: milvus
spec:
  dependencies: {}
  components: {}
  config: {}

Configurar NATS

NATS es un almacenamiento de mensajes alternativo para NATS.

Ejemplo

El siguiente ejemplo configura un servicio NATS.

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: milvus
spec:
  dependencies: 
    msgStreamType: 'natsmq'
    natsmq:
      # server side configuration for natsmq.
      server: 
        # 4222 by default, Port for nats server listening.
        port: 4222 
        # /var/lib/milvus/nats by default, directory to use for JetStream storage of nats.
        storeDir: /var/lib/milvus/nats 
        # (B) 16GB by default, Maximum size of the 'file' storage.
        maxFileStore: 17179869184 
        # (B) 8MB by default, Maximum number of bytes in a message payload.
        maxPayload: 8388608 
        # (B) 64MB by default, Maximum number of bytes buffered for a connection applies to client connections.
        maxPending: 67108864 
        # (√ms) 4s by default, waiting for initialization of natsmq finished.
        initializeTimeout: 4000 
        monitor:
          # false by default, If true enable debug log messages.
          debug: false 
          # true by default, If set to false, log without timestamps.
          logTime: true 
          # no log file by default, Log file path relative to.. .
          logFile: 
          # (B) 0, unlimited by default, Size in bytes after the log file rolls over to a new one.
          logSizeLimit: 0 
        retention:
          # (min) 3 days by default, Maximum age of any message in the P-channel.
          maxAge: 4320 
          # (B) None by default, How many bytes the single P-channel may contain. Removing oldest messages if the P-channel exceeds this size.
          maxBytes:
          # None by default, How many message the single P-channel may contain. Removing oldest messages if the P-channel exceeds this limit.    
          maxMsgs: 
  components: {}
  config: {}

Para migrar el almacenamiento de mensajes de RocksMQ a NATS, haga lo siguiente:

1. Detenga todas las operaciones DDL.

2. Llame a la API FlushAll y luego detenga Milvus una vez que la llamada a la API termine de ejecutarse.

3. Cambie msgStreamType a natsmq y realice los cambios necesarios en la configuración de NATS en spec.dependencies.natsmq.

4. Inicie Milvus de nuevo y compruebe si

   • Hay una entrada de registro con el nombre mqType=natsmq en los registros.
   • Hay un directorio llamado jetstream en el directorio especificado en spec.dependencies.natsmq.server.storeDir.

5. (Opcional) Haga una copia de seguridad y limpie los archivos de datos en el directorio de almacenamiento de RocksMQ.

Configurar Pulsar

Pulsar gestiona los registros de cambios recientes, genera registros de flujo y proporciona suscripciones a registros. Configurar Pulsar para el almacenamiento de mensajes está soportado tanto en Milvus standalone como en Milvus cluster. Sin embargo, con Milvus Operator, sólo puede configurar Pulsar como almacenamiento de mensajes para Milvus cluster. Añada los campos necesarios en spec.dependencies.pulsar para configurar Pulsar.

pulsar admite external y inCluster.

Pulsar externo

external indica que se utiliza un servicio Pulsar externo. Los campos utilizados para configurar un servicio Pulsar externo incluyen:

• external: Un valor true indica que Milvus utiliza un servicio Pulsar externo.
• endpoints: Los puntos finales de Pulsar.

Ejemplo

El siguiente ejemplo configura un servicio Pulsar externo.

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: my-release
  labels:
    app: milvus
spec:
  dependencies: # Optional
    pulsar: # Optional
      # Whether (=true) to use an existed external pulsar as specified in the field endpoints or 
      # (=false) create a new pulsar inside the same kubernetes cluster for milvus.
      external: true # Optional default=false
      # The external pulsar endpoints if external=true
      endpoints:
      - 192.168.1.1:6650
  components: {}
  config: {}           

Pulsar interno

inCluster indica que cuando se inicia un clúster Milvus, se inicia automáticamente un servicio Pulsar en el clúster.

Ejemplo

El siguiente ejemplo configura un servicio Pulsar interno.

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: my-release
  labels:
    app: milvus
spec:
  dependencies:
    pulsar:
      inCluster:
        values:
          components:
            autorecovery: false
          zookeeper:
            replicaCount: 1
          bookkeeper:
            replicaCount: 1
            resoureces:
              limit:
                cpu: '4'
              memory: 8Gi
            requests:
              cpu: 200m
              memory: 512Mi
          broker:
            replicaCount: 1
            configData:
              ## Enable `autoSkipNonRecoverableData` since bookkeeper is running
              ## without persistence
              autoSkipNonRecoverableData: "true"
              managedLedgerDefaultEnsembleSize: "1"
              managedLedgerDefaultWriteQuorum: "1"
              managedLedgerDefaultAckQuorum: "1"
          proxy:
            replicaCount: 1
  components: {}
  config: {}            

Suponiendo que el archivo de configuración se llama milvuscluster.yaml, ejecute el siguiente comando para aplicar la configuración.

kubectl apply -f milvuscluster.yaml

Configurar Kafka

Pulsar es el almacenamiento de mensajes predeterminado en un clúster Milvus. Si desea utilizar Kafka, añada el campo opcional msgStreamType para configurar Kafka.

kafka admite external y inCluster.

Kafka externo

external indica que se utiliza un servicio Kafka externo.

Los campos utilizados para configurar un servicio Kafka externo incluyen:

• external: Un valor true indica que Milvus utiliza un servicio Kafka externo.
• brokerList: La lista de brokers a los que enviar los mensajes.

Ejemplo

El siguiente ejemplo configura un servicio Kafka externo.

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: my-release
  labels:
    app: milvus
spec:
  config:
    kafka:
      # securityProtocol supports: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL 
      securityProtocol: PLAINTEXT
      # saslMechanisms supports: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
      saslMechanisms: PLAIN
      saslUsername: ""
      saslPassword: ""
  # Omit other fields ...
  dependencies:
    # Omit other fields ...
    msgStreamType: "kafka"
    kafka:
      external: true
      brokerList: 
        - "kafkaBrokerAddr1:9092"
        - "kafkaBrokerAddr2:9092"
        # ...

Las configuraciones SASL son compatibles con el operador v0.8.5 o versiones superiores.

Kafka interno

inCluster indica que cuando se inicia un cluster Milvus, se inicia automáticamente un servicio Kafka en el cluster.

Ejemplo

El siguiente ejemplo configura un servicio Kafka interno.

apiVersion: milvus.io/v1alpha1
kind: Milvus
metadata:
  name: my-release
  labels:
    app: milvus
spec: 
  dependencies:
    msgStreamType: "kafka"
    kafka:
      inCluster: 
        values: {} # values can be found in https://artifacthub.io/packages/helm/bitnami/kafka
  components: {}
  config: {}

Encuentre los elementos de configuración completos para configurar un servicio Kafka interno aquí. Añada los elementos de configuración que necesite en kafka.inCluster.values.

Suponiendo que el archivo de configuración se llama milvuscluster.yaml, ejecute el siguiente comando para aplicar la configuración.

kubectl apply -f milvuscluster.yaml

Lo que sigue

Aprenda a configurar otras dependencias de Milvus con Milvus Operator:

• Configure Object Storage con Milvus Operator
• Configure Meta Storage con Milvus Operator