Настройка server-side streaming#

Содержание#

Базовые ограничения и особенности
Настройка
Специфичные действия для цепочек обработки ответа
- Действие LazySessionTimeChecker
- Действие EventSourceConverter
Ограничение допустимого размера данных
Особенности применения стандартных действий
Примеры

Для настройки server-side streaming в профиле Platform V SOWA необходимо соблюдать определенную последовательность действий и учитывать те ограничения и возможности, которые предоставляются продуктом. Подробное описание приведено в статье ниже.

Базовые ограничения и особенности#

При использовании настроек, включающих поддержку серверного поточного взаимодействия в сервисе service_http_proxy, необходимо учесть следующее:

Стандартные параметры конфигурации, влияющие на обработку ответов, не допускаются к применению, либо работают иначе. А именно, параметры download_size_limit и filter_mds.download_size_limit - отсутствует при включении потоковой обработки ответов, filter_mds игнорирует тело ответа. Для выполнения соответствующей функциональности необходимо применять действия цепочек обработки ответов (ConditionChecker и FilterMds), далее будет сказано о них более подробно.
По умолчанию отключается буферизация ответа от бэкенда (в отличие от стандартной настройки service_http_proxy).
Существует одновременно два пути обработки ответов от бэкенда - поточных и непоточных.
Возможности стандартного механизма постпроцессора при обработке ответов ограничены в качестве используемых переменных и фазы ответа - его действие распространяется только на HTTP-status line и заголовки ответа. Все остальные операции по постобработке выполняются отдельным постпроцессором поточного ответа, чьи настройки и возможности отличаются от стандартного постпроцессора, и зависят, в том числе, от типа потока, который обрабатывается.
Логирование событий event_hook осуществляется в рамках сессии. Так как сессия может иметь достаточно большой объем закэшированных данных в памяти, требуется ограничивать объем выделяемой памяти для сохранения данных ответов (для сброса их на дисковое временное хранилище).

Настройка#

Базовые параметры#

Для включения поддержки потоковой передачи сообщений в сервисе service_http_proxy, минимально достаточно объявление следующей конструкции:

    server_streaming:
      stream_type: <ТИП СТРИМА>

Где <ТИП СТРИМА> - один из вариантов SSE, JSON_LINES, GRPC, PROTOBUF_HTTP.

Объявление такой конструкции позволяет, с одной стороны, сгенерировать базовые параметры для того, чтобы поток доходил до клиента, c другой стороны - определить параметры для парсера сообщений, который впоследствии используется в цепочках обработки ответа.

Важно! По умолчанию, выбранный тип потока определяет в том числе условия, по которым определяется является ли ответ поточным или обычным полным HTTP ответом. Для каждого типа потока определены следующие условия:

Тип потока	Условия
SSE	Заголовок HTTP Content-Type: ^(text/event-stream)(\+[^;]+)?(;.+)?$
GRPC	Заголовок HTTP Content-Type: ^(application/grpc)(\+[^;]+)?(;.+)?$
Protobuf HTTP	Заголовок HTTP Content-Type: ^(application/(x-protobuf\|protobuf\|vnd.google.protobuf))(;.+)?$
JSON Lines	Заголовок HTTP Content-Type: ^(?:application\|text)/(?:x-ndjson\|jsonl)(;.+)?$

Важно, чтобы условия соответствовали реальному потоку, так как в обратном случае, поток будет либо зависать, не доходя до клиента, либо завершатся с ошибкой. Для того чтобы переопределить условия, можно воспользоваться параметром parser_enable_condition.

Например,

    server_streaming:
      stream_type: SSE
      parser_enable_condition:
         - var: $sent_http_content_type
           val: ^(text/event-stream)(\\+[^;]+)?(;.+)?$
           operator: "~" 
         - var: $upstream_response_state
           operator: "<"
           val: 300  

Здесь для потока SSE событий переопределяются условия включения парсера по умолчанию, добавляя код статуса ответа меньше 300.

Для текстовых потоков (SSE/JSON LINES) также можно переопределить разделитель сообщений - по умолчанию, разделителем для SSE являются DOUBLE_LF (\n\n) и DOUBLE_CRLF (\r\n\r\n), для JSON Lines - LF (\n) и CRLF (\r\n). Можно явно указать разделитель, используя параметр message_delimiter. Для grpc и protobuf over HTTP всегда подразумевается использование length-prefix.

Еще одним параметром, который может повлиять на передачу событий по потокам SSE и JSON lines, является dont_chunk_response. По умолчанию, прокси-серверы (в том числе и SOWA) применяют chunked transfer-encoding для ответов, не содержащих Content-Length. Для того чтобы это предотвратить, можно использовать параметр dont_chunk_response, выставив его в true.

Настройка проверок и валидации#

Как уже говорилось ранее, вся функциональность проверки поточных данных реализуется посредством цепочек обработки событий.

Цепочки обработки событий ответов при включенной поддержке потоковой обработки подразделяются на две категории - те, которые обрабатывают только заголовки ответа/ответы с полным сообщением и цепочки, которые обрабатывают потоковые сообщения. В какие из этих цепочек попадет ответ, определяется параметром parser_enable_condition (или параметрами парсера по умолчанию). Это сделано из-за того, что при взаимодействии с бэкендом можно получить ошибочный ответ (например, 502 Bad Gateway), который может не перехватываться постпроцессором, и при этом не является потоковым SSE/GRPC и т.п. сообщением.

Для того чтобы пометить цепочку как потоковую, используется параметр streaming при определении цепочки, например:

      response_chains:
      - actions:
        - action: ConditionChecker
          params:
            rules:
            - var: $upstream_response_state
              val: 400
              operator: <
      - message: $clj_response_body
        streaming: true
        actions:
        - action: CheckDataSize
          params:
            max_allowable_size: '1024'

Здесь две цепочки:

первая - не поточная, отрабатывает на заголовках ответа;
вторая - поточная, в ней выставлен флаг streaming в значение true.

Важно

Все стримминговые цепочки должны содержать в параметре message сообщение clj_request_body/$clj_response_body.

Постпроцессинг сообщений#

При обработке потоковых сообщений нужно иметь в виду, что повлиять на HTTP статус ответа и заголовка ответа, основываясь на теле сообщений не получится - они отправляются клиенту до того, как начинается получение и обработка непосредственно самих сообщений.

Возможная реакция на негативный результат обработки сообщений действиями в цепочке определяется как типом потока, так и параметрами постпроцессора потоковых сообщений, если он определен. Если не определен, то реакция одна - разрыв соединения.

Параметры постпроцессора потоковых сообщений приведены в разделе Постобработчики, здесь даны некоторые пояснения и общая информация по использованию.

Объявление#

Объявление производится через задание параметра message_error_processor в сервисе service_http_proxy:

  message_error_processor:
    rules_file: message_rules_sse.yml

rules_file является единственным обязательным параметром и содержит ссылку на файл с правилами постобработки, который традиционно должен располагаться в директории с кастомными ресурсами профиля. Остальные параметры (как и содержимое файла с правилами) зависят от типа потока, поэтому они будут рассмотрены отдельно.

Постобработка сообщений SSE/JSON LINES#

В случае stream_type - SSE/JSON LINES, в параметре message_error_processor можно дополнительно объявить вложенный параметр pattern_file. Данный параметр содержит шаблон сообщения, отправляемого клиенту в случае ошибки/блокировки потокового сообщения, если в правилах предусмотрена отправка такого сообщения. Сам файл должен содержать все разделители между полями event-source (в случае SSE) события, включая последовательность символов окончания сообщения, например:

event: error\r\ndata: incorrect data format in {request_id} request\n\n

Правила#

Файл с правилами должен иметь как минимум один параметр action, определяющий, какое действие будет выполняться при соблюдении условий этого правила. В случае, если ни одно действие не подходит по своим условиям, применяется правило с указанным параметром default: true.

Возможные действия:

SEND_MSG - отправить сконструированное сообщение вместо заблокированного/вызвавшего ошибку;
DROP_MSG - удалить сообщение из очереди на отправку;
SEND_MSG_CLOSE - отправить сообщение и корректно завершить соединение с клиентом;
CLOSE - корректно завершить соединение с клиентом;
DISCONNECT - разорвать соединение с клиентом.

В случае, если выбрано действие SEND_MSG или SEND_MSG_CLOSE, дополнительно можно указать параметры responsePattern или responsePatternFile, содержащий шаблон сообщения в виде строки или в виде ссылки на файл и replacements - набора соответствий вида <текст из шаблона, отмеченный в шаблоне фигурными скобками> и <текст с переменными для замены>.

Пример:

  - action: SEND_MSG
    priority: 0
    responsePattern: "event: error\r\ndata: incorrect data format in {request_id} request\n\n"
    replacements:
      - pattern: request_id
        val: $request_id
    condition: 
       - var: clj_error_action
         val: EventSourceConverter
         operator: '='

Остальные параметры работают аналогично традиционному постпроцессору.

Важно

responsePattern/responsePatternFile переопределяют pattern_file, указанный в message_error_processor;
только одно правило может иметь признак default, установленный в true.

Постобработка сообщений GRPC#

Постобработчик потоковых сообщений GRPC отличается от SSE тем, что вместо pattern_file определяется параметр proto_config, содержащий ссылку на proto файл шаблона ответного сообщения в виде:

  proto_config:
    # всегда proto_file
    type: proto_file
    # ссылка на proto файл
    path: schemes/proto_file/monitor.proto
    # java class namr
    class: monitor.Data
    # Сообщение
    message: StatUpdate
    # всегда data
    generate: data

Важно

Стоит учитывать, что возможности по замене подстановок ограничены, поэтому стоит использовать простые типы данных.

Правила#

Кроме действий, доступных для http стриминга здесь появляются дополнительные SEND_TRAILERS и SEND_MSG_TRAILERS, позволяющие отправить завершающие трейлеры и сообщение с завершающими трейлерами после него. И в том и другом случае, после отправки соединение закрывается.

Вместо responsePattern/responsePatternFile используется параметр protoConfig.

В качестве replacements указывается сочетание fieldName - имя поля в message классе и val - значение (возможно содержащее переменные для подстановки).

Пример:

 rules:
 - action: SEND_MSG
   priority: 100
   protoConfig: {path: schemes/proto_file/user.proto, class: Protobuf.Two, message: User}
   replacements:
     - fieldName: request_id
       val: $request_id
   condition: 
     - var: clj_error_action
       val: HttpFilter
       operator: '='rules:

При заполнении trailers, используется соответствующий параметр trailers.

Пример:

rules:
- action: SEND_MSG_TRAILERS
  priority: 100
  protoConfig: {path: schemes/proto_file/user.proto, class: Protobuf.Two, message: User}
  trailers:
    - name: grpc-status
      val:  403
    - name: grpc-message
      val:  response for $request_id was blocked for some reason
  replacements:
    - fieldName: request_id
      val: $request_id
  condition: 
    - var: clj_error_action
      val: HttpFilter
      operator: '='rules:

Специфичные действия для цепочек обработки ответа#

Действие LazySessionTimeChecker#

Действие предназначено для «ленивой» проверки максимального времени жизни пользовательской сессии и ее принудительному разрыву по достижению ограничения. Особенно полезно для HTTP-стриминга, где не предусмострено heartbit клиента, и при некорректном закрытии tcp соединения они могут оставаться открытыми на стороне шлюза до истечения времени keep-alive tcp.

Действие имеет единственный параметр sessionTime, в котором указывается время жизни сессии.

Действия считается «ленивым» по той причине, что проверка времени существования сессии осуществляется при получении очередной порции данных от бэкенда.

Действие EventSourceConverter#

Конвертирует event-source событие, передаваемое со стороны SSE-сервера в JSON/XML объект. Полезно при последующей фильтрации данных с использованием Modsecurity, либо валидации на соответствие схеме валидации.

Имеет следующие параметры:

outputFormat - позволяет выбрать JSON или XML объект получать на выходе;
dataMode - при использовании многострочных элементов data, позволяет их либо собирать в одну строку, либо представлять как массив строк;
commentMode - аналогично для комментариев;
convertData - конвертировать ли элемент data в json/xml объект или представлять его как строку;
convertComment - аналогично для комментариев;
includeElements - список элементов (event/id/retry/data/comment), которые будут включены в результирующий объект;
excludeElements - список элементов (event/id/retry/data/comment), которые будут исключены из результирующего объекта.

Типичный пример:

 - action: EventSourceConverter
   params:
     outputFormat: JSON
     convertData: true
     dataMode: STRING
     excludeElements: 
      - comment

Ограничение допустимого размера данных#

Для ограничения на максимальный размер данных полученных от бэкенда (аналог download_size_limit) используйте переменную $clj_stream_message_size в действии ConditionChecker.

Для ограничения размера индивидуального сообщения используйте действие CheckDataSize или переменную clj_message_size.

Особенности применения стандартных действий#

при использовании действия FilterMds учитывайте тот факт, что данные из сообщения отправляются как запрос;
при использовании действий ProtobufToJson и ProtobufValidation учитывайте, что в эти действия Protobuf сообщение попадает уже без zip флага и префикса с длиной, поэтому в них не нужно выставлять параметры isGrpc и merge_type.

Примеры#

SSE streaming#

Конфигурационный файл сервиса:

service_http_proxy
  - id: sse
    name: sse
    allowed_queries:
    - method: get
    server_streaming:
      stream_type: SSE
      parser_enable_condition:
         - var: $sent_http_content_type
           val: ^(text/event-stream)(\\+[^;]+)?(;.+)?$
           operator: "~" 
         - var: $upstream_response_state
           operator: "<"
           val: 300  
    filters:
      filter_mds:
        enabled: true
        rules:
        - sowa-http
    chains:
      request_chains:
      - actions:
        - action: JsonValidation
          conditions:
          - operator: '>'
            val: 0
            var: $clj_message_size
          params:
            validation_schema:
              path: json-schema.json
              type: file
        message: $clj_request_body
      response_chains:
      - actions:
        - action: ConditionChecker
          params:
            rules:
            - var: $upstream_response_state
              val: 400
              operator: <
      - message: $clj_response_body
        streaming: true
        actions:
        - action: CheckDataSize
          params:
            max_allowable_size: '1024'
        - action: LazySessionTimeChecker
          params:
            sessionTime: 600s
        - action: ConditionChecker
          params:
            rules:
            - operator: <=
              var: $clj_stream_message_size
              val: 10000000
        - action: EventSourceConverter
          params:
            outputFormat: JSON
            convertData: true
            dataMode: STRING
          actions:
          - action: JsonValidation
            conditions:
            - operator: '>'
              val: 0
              var: $clj_message_size
            - operator: '='
              val: 200
              var: $upstream_status
            params:
              action: deny
              validation_schema: /sowa/profile_storage/custom/valid_test/json-schema.json
            id: a82b6407-cd82-4858-a733-e619d1588fab
          - action: FilterMds
            params:
              headers:
              - name: Content-Type
                value: application/json
              filter_mds:
                enabled: true
                ignore_parent: true
                rules:
                 - sowa-http-reversed
      message_error_processor:
        rules_file: message_rules.yml
    optional:
      event_hook:
          log_in_file: on
          enabled: true
          stage: 
            - name: client_request
            - name: upstream_response
            - name: client_response
      upload_size_limit:
        size: 300M
        drop: off
      http_version: 1.1
    upstream_group_id: static
    upstream_groups:
    - id: static
      rewrite_uri:
        - from: /sse
          to:  /
      servers:
      - server: 127.0.0.1:4012
        weight: 1
    url: ^/sse

Правила обработки ошибок сообщений(message_rules.yml):

rules:
  - action: SEND_MSG
    responsePattern: ":message blocked by policy\n\n"
    priority: 100
    condition: 
       - var: clj_error_action
         val: HttpFilter
         operator: '='
  - action: SEND_MSG
    priority: 0
    responsePattern: "event: error\r\ndata: incorrect data format in {request_id} request\n\n"
    replacements:
      - pattern: "request_id"
        val: $request_id
    condition: 
       - var: clj_error_action
         val: EventSourceConverter
         operator: '='
  - action: DROP_MSG
    priority: 0 
    responsePattern: "event: error\r\ndata: messsage too big for {request_id}\n\n"
    replacements:
      - pattern: "request_id"
        val: $request_id
    condition: 
       - var: clj_error_action
         val: CheckDataSize
         operator: '='
  - action: SEND_MSG_CLOSE
    priority: 0
    responsePattern: "event: error\r\ndata: max stream size exceeded\n\n"
    condition: 
       - var: clj_error_action
         val: ConditionChecker
         operator: '='
  - action: CLOSE
    priority: 0 
    condition: 
       - var: clj_error_action
         val: LazySessionTimeChecker
         operator: '='       

GRPC streaming#

Конфигурационный файл сервиса:

  service_http_proxy:
  - allowed_queries:
    - method: get
    - method: post
    id: grpc
    name: grpc
    server_streaming:
      stream_type: GRPC
    chains:
      response_chains:
      - actions:
        - action: ConditionChecker
          params:
            rules:
            - var: $upstream_response_state
              val: 400
              operator: <
      - message: $clj_response_body
        streaming: true
        actions:
        - action: CheckDataSize
          params:
            max_allowable_size: '1024'
        - action: LazySessionTimeChecker
          params:
            sessionTime: 30s
        - action: ConditionChecker
          params:
            rules:
            - operator: <=
              var: $clj_stream_message_size
              val: 10000000
        - action: ProtobufToJson
          params:
            proto_config:
              type: proto_file
              path: schemes/proto_file/monitor.proto
              class: monitor.Data
              message: StatUpdate
          actions:
          - action: JsonValidation
            conditions:
            - operator: '>'
              val: 0
              var: $clj_message_size
            - operator: '='
              val: 200
              var: $upstream_status
            params:
              action: deny
              validation_schema: /sowa/profile_storage/custom/valid_test/json-schema.json
            id: a82b6407-cd82-4858-a733-e619d1588fab
          - action: FilterMds
            params:
              headers:
              - name: Content-Type
                value: application/json
              filter_mds:
                enabled: true
      message_error_proccessor:
        rules_file: message_rules_grpc.yml
        proto_config:
          type: proto_file
          path: schemes/proto_file/monitor.proto
          class: monitor.Data
          message: StatUpdate
          generate: data
    optional:
      upload_size_limit:
        size: 300M
        drop: off
      http_version: 2.0
    upstream_group_id: static
    upstream_groups:
    - id: static  
      servers:
      - server: 127.0.0.1:50051
        weight: 1
    url: ^/monitor.MonitorService/StreamStats

Файл с правилами постобработки сообщений (message_rules_grpc.yml)

rules:
  - action: SEND_MSG
    priority: 100
    protoConfig: {path: schemes/proto_file/user.proto, class: Protobuf.Two, message: User}
    replacements:
      - fieldName: request_id
        val: $request_id
    condition: 
       - var: clj_error_action
         val: HttpFilter
         operator: '='
  - action: SEND_MSG
    priority: 0
    replacements:
      - fieldName: request_id
        val: $request_id
    condition: 
       - var: clj_error_action
         val: JsonValidation
         operator: '='
  - action: DROP_MSG
    priority: 0 
    replacements:
      - fieldName: request_id
        val: $request_id
    condition: 
       - var: clj_error_action
         val: CheckDataSize
         operator: '='
  - action: SEND_MSG_CLOSE
    priority: 0
    condition: 
       - var: clj_error_action
         val: ConditionChecker
         operator: '='
  - action: CLOSE
    priority: 0 
    condition: 
       - var: clj_error_action
         val: LazySessionTimeChecker
         operator: '='       

JSON lines#

Конфигурационный файл сервиса:

  - allowed_queries:
    - method: get
    - method: post
    id: http_lines
    listen:
    - port: 4045
    name: json_lines
    server_streaming:
      stream_type: JSON_LINES
    filters:
      filter_mds:
        enabled: true
        rules:
        - sowa-http
    chains:
      response_chains:
      - actions:
        - action: ConditionChecker
          params:
            rules:
            - var: $upstream_response_state
              val: 400
              operator: <
      - message: $clj_response_body
        actions:
        - action: JsonValidation
          conditions:
          - val: 300
            var: $status
            operator: <
          - operator: '='
            val: application/json
            var: $http_content_Type
          - operator: '>'
            val: 0
            var: $clj_message_size
          params:
            validation_schema:
              path: json-schema.json
              type: file
        - action: CheckDataSize
          conditions:
          - val: 300
            var: $status
            operator: <        
          params:
            max_allowable_size: '1024'
        - action: LazySessionTimeChecker
          conditions:
          - val: 300
            var: $status
            operator: <        
          params:
            sessionTime: 600s
        - action: ConditionChecker
          conditions:
          - val: 300
            var: $status
            operator: <
          params:
            rules:
            - operator: <=
              var: $clj_stream_message_size
              val: 10000000
        - action: FilterMds
          conditions:
          - val: 300
            var: $status
            operator: <        
          params:
            headers:
             - name: Content-Type
               value: application/json
            filter_mds:
              enabled: true
      message_error_processor:
        rules_file: message_rules_http_lines.yml
    optional:
      upload_size_limit:
        size: 300M
        drop: off
      http_version: 1.1
    upstream_group_id: static
    upstream_groups:
    - id: static
      servers:
      - server: 127.0.0.1:4022
        weight: 1
    url: ^/json

Файл с правилами постобработки сообщений (message_rules_http_lines.yml) аналогичен тому, что показан для SSE.